如何使用 CZMQ-4.0.2 新的 zsock API 创建发布/订阅架构？

Question

我想使用 CZMQ-4.0.2 创建一个发布/订阅架构，但我不明白如何使用新的 zsock API。

谁能指出一些使用新 API 的示例？

Answer 1

tldr;

Examples are on the bottom of the site

小解释

我假设您询问的是 CZMQ 的特定用法，而不是如何使用 ZeroMQ 套接字，以及 PUB/SUB 模式的怪癖是什么。

使用 CZMQ 时，您无需担心上下文，它是在内部完成的。 zsock_new 函数族返回指向 zsock_t 的指针，这是套接字的不透明标识符。完成后请记得致电zsock_destroy(&socket)，以避免内存泄漏。

在最常见的用法中，您无需担心连接和绑定，因为 zsock_new_XXX 会处理这些问题。要了解采取了哪些措施，您可以查看manual。

//  Create a PUB socket. Default action is bind.
CZMQ_EXPORT zsock_t *
    zsock_new_pub (const char *endpoint);
//  Create a SUB socket, and optionally subscribe to some prefix string. Default
//  action is connect.
CZMQ_EXPORT zsock_t *
    zsock_new_sub (const char *endpoint, const char *subscribe);

如果您打算进行一些不寻常的绑定/连接，您可以在endpoint 中添加前缀。 @ 表示绑定，> 连接。

zsock_t *sock = zsock_new_push("@ipc://test");

现在，要发送消息，您可以使用多种方法（zsock_send、zmsg_send、zstr_send、zstr_sendx、zstr_sendf、zframe_send），最通用的是zsock_send。它具有类似原型的 printf，您需要在其中传递消息的图片。此字符串中的每个字符都代表消息中的单个帧（或更多帧，因为您还可以传递另一条消息）。在here中有描述：

//  Send a 'picture' message to the socket (or actor). The picture is a
//  string that defines the type of each frame. This makes it easy to send
//  a complex multiframe message in one call. The picture can contain any
//  of these characters, each corresponding to one or two arguments:
//
//      i = int (signed)
//      1 = uint8_t
//      2 = uint16_t
//      4 = uint32_t
//      8 = uint64_t
//      s = char *
//      b = byte *, size_t (2 arguments)
//      c = zchunk_t *
//      f = zframe_t *
//      h = zhashx_t *
//      U = zuuid_t *
//      p = void * (sends the pointer value, only meaningful over inproc)
//      m = zmsg_t * (sends all frames in the zmsg)
//      z = sends zero-sized frame (0 arguments)
//      u = uint (deprecated)
//
//  Note that s, b, c, and f are encoded the same way and the choice is
//  offered as a convenience to the sender, which may or may not already
//  have data in a zchunk or zframe. Does not change or take ownership of
//  any arguments. Returns 0 if successful, -1 if sending failed for any
//  reason.
CZMQ_EXPORT int
zsock_send (void *self, const char *picture, ...);

一个可能不清楚的是这个void *self，它实际上是我们从zsock_new返回的zsock_t *。在原型中，它被声明为void *，因为这个函数也接受zactor_t *。

重要提示：Does not change or take ownership of any arguments.。您需要在发送后释放/销毁数据。

接收看起来非常相似。就像 sscanf 一样，zsock_recv 创建对象，所以同样需要注意内存。

ZeroMQ 和 CZMQ 在行为上的最大区别在于 LINGER 套接字选项。对于 ZeroMQ，它是无限的 (-1)，其中 CZMQ 的默认值为 0（无阻塞）。因此，任何时候当您有zsock_send 后跟zsock_destroy 时，您的消息可能无法传递。可以使用zsock_set_linger 为套接字单独设置Linger 值，也可以使用全局zsys_set_linger 设置。

发布者示例

#include <czmq.h>

int main(int argc, char ** argv) {
  zsock_t *socket = zsock_new_pub("ipc://example.sock");
  assert(socket);

  while(!zsys_interrupted) {
    zsys_info("Publishing");
    zsock_send(socket, "sss", "TOPIC", "MESSAGE PART", "ANOTHER");
    zclock_sleep(1000);
  }

  zsock_destroy(&socket);
  return 0;
}

订阅者示例

#include <czmq.h>

int main(int argc, char ** argv) {
  zsock_t *socket = zsock_new_sub("ipc://example.sock", "TOPIC");
  assert(socket);

  char *topic;
  char *frame;
  zmsg_t *msg;
  int rc = zsock_recv(socket, "sm", &topic, &msg);
  assert(rc == 0);

  zsys_info("Recv on %s", topic);
  while(frame = zmsg_popstr(msg)) {
    zsys_info("> %s", frame);
    free(frame);
  }
  free(topic);
  zmsg_destroy(&msg);

  zsock_destroy(&socket);
  return 0;
}