C/C++编程：ZeroMQ API学习

╰+攻爆jí腚メ 2022-11-06 15:55 478阅读 0赞

> 前文：[C/C++编程：ZeroMQ安装以及使用（windows+centos）][C_C_ZeroMQ_windows_centos]

>     #include <zmq.h>
> 
>  *  cc \[flags\] files **`-lzmq`** \[libraries\]

### 文章目录 ###

*  版本
 *   *  zmq\_version()
 *  上下文
 *   *  zmq\_ctx\_new
     *  zmq\_ctx\_set
     *  zmq\_ctx\_get
     *  zmq\_ctx\_term
     *  示例
 *  消息
 *   *  zmq\_msg\_init
     *  zmq\_msg\_init\_size
     *  zmq\_msg\_init\_data
     *  zmq\_msg\_send
     *  zmq\_msg\_recv
     *  zmq\_msg\_close
 *  socket
 *   *  zmq\_socket
     *  zmq\_setsockopt

# 版本 #

## zmq\_version() ##

>     void zmq_version（int * major，int * minor，int * patch）;
> 
>  *  ZeroMQ API参考手册：[http://api.zeromq.org/master:zmq-version][http_api.zeromq.org_master_zmq-version]
>  *  **功能**： 获取当前使用的ZeroMQ的版本号
>  *  **参数**：
>     
>      *  1.主版本号
>      *  2.次要版本号
>      *  3.补丁号

> **示例**：

//version.c
    #include <stdio.h>
    #include <zmq.h>
     
    int main()
    {
        
        int major, minor, patch;
        zmq_version(&major, &minor, &patch);
        printf("Current ZeroMQ version is %d.%d.%d\n", major, minor, patch);
    }

![在这里插入图片描述][2021031014244513.png]

# 上下文 #

ZMQ上下文保留套接字列表并管理异步IO线程和内部查询

*  在使用任何ØMQ库函数之前，必须创建一个ØMQ上下文。
 *  退出应用程序时，必须销毁上下文。

这些功能使您可以处理上下文：

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>作用</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>zmq_ctx_new</td> 
   <td>创建一个新的ØMQ上下文</td> 
  </tr> 
  <tr> 
   <td>zmq_ctx_set</td> 
   <td>设置上下文选项</td> 
  </tr> 
  <tr> 
   <td>zmq_ctx_get</td> 
   <td>获取上下文选项</td> 
  </tr> 
  <tr> 
   <td>zmq_ctx_shutdown</td> 
   <td>关闭ØMQ上下文</td> 
  </tr> 
  <tr> 
   <td>zmq_ctx_term</td> 
   <td>终止上下文</td> 
  </tr> 
 </tbody> 
</table>

## zmq\_ctx\_new ##

>     void *zmq_ctx_new ();
> 
>  *  ZeroMQ API参考手册：[http://api.zeromq.org/master:zmq-ctx-new][http_api.zeromq.org_master_zmq-ctx-new]
>  *  **功能**：
>     
>      *  创建新的ØMQ上下文
>      *  该函数代替了不推荐使用的函数`zmq_init（3）`
>      *  ØMQ上下文是线程安全的，可以根据需要在尽可能多的应用程序线程之间共享，而调用方无需进行任何其他锁定。
>  *  **参数**：
>     
>      *  无
>  *  **返回值**：
>     
>      *  如果成功，zmq\_ctx\_new（）函数应向新创建的上下文返回一个不透明的句柄。
>      *  否则，它将返回NULL并将errno设置为以下定义的值之一。
>  *  **错误**：
>     
>      *  没有为该功能定义错误值。

## zmq\_ctx\_set ##

>     int zmq_ctx_set（void * context，int option_name，int option_value）;
> 
>  *  ZeroMQ API参考手册：[http://api.zeromq.org/master:zmq-ctx-set][http_api.zeromq.org_master_zmq-ctx-set]
>  *  **功能**：
>     
>      *  zmq\_ctx\_set()函数将option\_name参数指定的选项设置为option\_value参数的值。
>  *  zmq\_ctx\_set()函数接受如下**选项**:：
>     
>      *  **ZMQ\_BLOCKY**：
>      *  **ZMQ\_IO\_THREADS**：设置I / O线程数
>      *  **ZMQ\_THREAD\_SCHED\_POLICY**：设置I / O线程的调度策略
>      *  **ZMQ\_THREAD\_PRIORITY**：设置I / O线程的调度优先级
>      *  **ZMQ\_THREAD\_AFFINITY\_CPU\_ADD**：将CPU添加到I / O线程的相似性列表
>      *  **ZMQ\_THREAD\_AFFINITY\_CPU\_REMOVE**：从CPU移除对I / O线程的相似性列表
>      *  **ZMQ\_THREAD\_NAME\_PREFIX**：设置I / O线程的名称前缀
>      *  **ZMQ\_MAX\_MSGSZ**：设置最大消息大小
>      *  **ZMQ\_ZERO\_COPY\_RCV**：指定消息解码策略
>      *  **ZMQ\_MAX\_SOCKETS**：设置最大套接字数
>      *  **ZMQ\_IPV6**：设置IPv6选项
>  *  **返回值**：
>     
>      *  如果成功，zmq\_ctx\_set（）函数将返回零
>      *  否则，它将返回-1并将errno设置为以下定义的值之一。
>  *  **错误**：
>     
>      *  EINVAL：请求的选项option\_name未知。

## zmq\_ctx\_get ##

>     int zmq_ctx_get（void * context，int option_name）;
> 
>  *  ZeroMQ API参考手册：[http://api.zeromq.org/master:zmq-ctx-get][http_api.zeromq.org_master_zmq-ctx-get]
>  *  **功能**：
>     
>      *  返回由指定的选项option\_name的参数。
>  *  zmq\_ctx\_get()函数接受如下**选项**:：
>     
>      *  **ZMQ\_IO\_THREADS**：获取I / O线程数
>      *  **ZMQ\_MAX\_SOCKETS**：获取最大套接字数
>      *  **ZMQ\_MAX\_MSGSZ**：获取最大消息大小
>      *  **ZMQ\_ZERO\_COPY\_RCV**：获取消息解码策略
>      *  **ZMQ\_SOCKET\_LIMIT**：获得最大数量的可配置套接字
>      *  **ZMQ\_IPV6**：设置的IPv6选项
>      *  **ZMQ\_BLOCKY**：获取阻塞设置
>      *  **ZMQ\_THREAD\_SCHED\_POLICY**：获取I / O线程的调度策略
>      *  **ZMQ\_THREAD\_NAME\_PREFIX**：获取I / O线程的名称前缀
>      *  **ZMQ\_MSG\_T\_SIZE**：在运行时获取zmq\_msg\_t的大小

## zmq\_ctx\_term ##

>     int zmq_ctx_term（void * context）
> 
>  *  ZeroMQ API参考手册：[http://api.zeromq.org/master:zmq-ctx-term][http_api.zeromq.org_master_zmq-ctx-term]
>  *  **功能**：
>     
>      *  终止上下文。上下文终止按以下步骤执行：
>         
>          *  当前在上下文中打开的套接字上正在运行的任何阻塞都应该立即返回，并返回错误码`ETERM`。除zmq\_close（）之外，在上下文中打开的套接字上的任何其他操作均将失败，并显示错误代码ETERM。
>          *  在中断所有阻塞调用之后，zmq\_ctx\_term（）将阻塞，直到满足以下条件：
>             
>              *  在上下文中打开的所有套接字都已使用zmq\_close（）关闭。
>              *  对于上下文中的每个套接字，应用程序使用zmq\_send（）发送的所有消息都已物理传输到网络对等方，或者使用ZMQ\_LINGER套接字选项设置的套接字停留时间已到期。
>      *  该函数替换了不推荐使用的函数`zmq_term（3）`和`zmq_ctx_destroy（3）`。
>  *  **返回值**：
>     
>      *  如果成功，zmq\_ctx\_term（）函数应返回零。
>      *  否则，它将返回-1并将errno设置为以下定义的值之一。
>  *  **错误**：
>     
>      *  EFAULT：提供的上下文无效。
>      *  EINTR：终止被信号中断。如果需要，可以重新启动它。

## 示例 ##

> **示例**：

*  设置套接字的数量限制

//version.c
    #include <stdio.h>
    #include <zmq.h>
    #include <assert.h>
    
    int main()
    {
        
        void *context = zmq_ctx_new ();
        zmq_ctx_set (context, ZMQ_MAX_SOCKETS, 256);
        int max_sockets = zmq_ctx_get (context, ZMQ_MAX_SOCKETS);
        assert (max_sockets == 256);
    }

*  关闭上下文死锁策略

zmq_ctx_set (ctx, ZMQ_BLOCKY, false);

# 消息 #

*  ZMQ消息是在应用程序或者同一应用程序的组件之间传递的离散数据单元
 *  ZMQ消息没有内部结构，从ZMQ本身的角度来看，它们被视为不透明的二进制数据

<table> 
 <thead> 
  <tr> 
   <th>API</th> 
   <th>作用</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>zmq_msg_init</td> 
   <td>初始化空ØMQ消息</td> 
  </tr> 
  <tr> 
   <td>zmq_msg_init_size</td> 
   <td>初始化指定大小的ØMQ消息</td> 
  </tr> 
  <tr> 
   <td>zmq_msg_init_data</td> 
   <td>从提供的缓冲区初始化ØMQ消息</td> 
  </tr> 
  <tr> 
   <td>zmq_msg_send</td> 
   <td>发送消息</td> 
  </tr> 
  <tr> 
   <td>zmq_msg_recv</td> 
   <td>接收消息</td> 
  </tr> 
  <tr> 
   <td>zmq_msg_close</td> 
   <td>请求释放消息</td> 
  </tr> 
 </tbody> 
</table>

## zmq\_msg\_init ##

>     int zmq_msg_init（zmq_msg_t * msg）
> 
>  *  **功能**：
>     
>      *  `zmq_msg_init()`函数将初始化msg引用的消息对象，以表示一个空消息。
>      *  这个函数在使用`zmq_msg_recv()`接收消息之前调用时最有用。
>      *  永远不要直接访问zmq\_msg\_t成员，而是始终使用zmq\_msg函数族。
>      *  函数zmq\_msg\_init()、zmq\_msg\_init\_data()和zmq\_msg\_init\_size()是互斥的。永远不要初始化相同的zmq\_msg\_t两次。
>  *  **返回值**：
>     
>      *  该zmq\_msg\_init（）函数总是返回零。

> **示例**：

*  从套接字接收消息

zmq_msg_t msg;
        rc = zmq_msg_init (&msg);
        assert (rc == 0);
        int nbytes = zmq_msg_recv (socket, &msg, 0); 
        assert (nbytes != -1);

## zmq\_msg\_init\_size ##

>     int zmq_msg_init_size (zmq_msg_t *msg, size_t size);
> 
>  *  **功能**：
>     
>      *  zmq\_msg\_init\_size()函数将分配任何存储消息大小字节长度所需的资源，并初始化msg引用的消息对象来表示新分配的消息。
>      *  实现应该选择是将消息内容存储在堆栈(小消息)还是堆(大消息)上。由于性能原因zmq\_msg\_init\_size()不应该清除消息数据。
>      *  永远不要直接访问zmq\_msg\_t成员，而是始终使用zmq\_msg函数族。
>      *  函数zmq\_msg\_init()、zmq\_msg\_init\_data()和zmq\_msg\_init\_size()是互斥的。永远不要初始化相同的zmq\_msg\_t两次。
>  *  **返回值**：
>     
>      *  如果成功，zmq\_msg\_init\_size（）函数应返回零。
>      *  否则，它将返回-1并将errno设置为以下定义的值之一。
>  *  **错误**：
>     
>      *  ENOMEM：没有足够的存储空间。

## zmq\_msg\_init\_data ##

>     typedef void (zmq_free_fn) (void *data, void *hint);
>     int zmq_msg_init_data (zmq_msg_t *msg, void *data, size_t size, zmq_free_fn *ffn, void *hint);
> 
>  *  **功能**：
>     
>      *  zmq\_msg\_init\_data()函数将初始化msg引用的消息对象，以表示地址数据处的缓冲区引用的内容，大小为字节长。不得复制数据，且ØMQ应获得所提供缓冲区的所有权。
>      *  如果提供了释放功能，则一旦ØMQ不再需要数据缓冲区，就应调用释放函数ffn，并将数据和提示参数提供给zmq\_msg\_init\_data（）。
>      *  如果未提供释放功能，则不会释放已分配的内存，这可能会导致内存泄漏。
>      *  释放函数ffn必须是线程安全的，因为它将从任意线程调用。
>      *  永远不要直接访问zmq\_msg\_t成员，而是始终使用zmq\_msg函数族。
>      *  函数zmq\_msg\_init()、zmq\_msg\_init\_data()和zmq\_msg\_init\_size()是互斥的。永远不要初始化相同的zmq\_msg\_t两次。
>  *  **返回值**：
>     
>      *  如果成功，zmq\_msg\_init\_data（）函数应返回零。
>      *  否则，它将返回-1并将errno设置为以下定义的值之一。
>  *  **错误**：
>     
>      *  ENOMEM：没有足够的存储空间。

> **示例**：

*  从提供的缓冲区初始化消息

void my_free (void *data, void *hint)
    	{
        
    	 free (data);
    	}
    
     /* ... */
    
    	void *data = malloc (6);
    	assert (data);
    	memcpy (data, "ABCDEF", 6);
    	zmq_msg_t msg;
    	rc = zmq_msg_init_data (&msg, data, 6, my_free, NULL);
    	 assert (rc == 0);

## zmq\_msg\_send ##

>     int zmq_msg_send (zmq_msg_t *msg, void *socket, int flags);
> 
>  *  **功能**：
>     
>      *  在套接字上发送消息部分

## zmq\_msg\_recv ##

>     int zmq_msg_recv（zmq_msg_t * msg，void * socket，int flags）;
> 
>  *  **功能**：
>     
>      *  从套接字接收消息部分

## zmq\_msg\_close ##

>     int zmq_msg_close (zmq_msg_t *msg);
> 
>  *  **功能**：
>     
>      *  zmq\_msg\_close 将通过ZMQ基础设施，与msg引用的消息对象相关联的所有资源不被需要，要求释放这些资源
>      *  与消息对象关联的资源的实际释放应该由ØMQ延迟，直到消息或底层数据缓冲区的所有用户都表示不再需要它。
>      *  应用程序应该确保在不再需要消息时调用zmq\_msg\_close()，否则可能会发生内存泄漏。注意，在zmq\_msg\_send()成功之后就没有必要这样做了。
>      *  永远不要直接访问zmq\_msg\_t成员，而是始终使用zmq\_msg函数族。
>  *  **返回值**：
>     
>      *  如果成功，zmq\_ctx\_set（）函数将返回零
>      *  否则，它将返回-1并将errno设置为以下定义的值之一。
>  *  **错误**：
>     
>      *  EFAULT：无效的消息

# socket #

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>zmq_socket</td> 
   <td>创建一个套接字</td> 
  </tr> 
  <tr> 
   <td>zmq_close</td> 
   <td>关闭套接字</td> 
  </tr> 
 </tbody> 
</table>

## zmq\_socket ##

>     void *zmq_socket (void *context, int type);
> 
>  *  ZeroMQ API参考手册：[http://api.zeromq.org/master:zmq-socket][http_api.zeromq.org_master_zmq-socket]
>  *  **功能**：
>     
>      *  zmq\_socket函数将在指定的上下文中创建一个ZMQ套接字，并为新创建的套接字返回一个不透明的句柄。
>      *  type参数指定套接字类型，它决定着套接字上通信的语义
>      *  新创建的套接字最初是未绑定的，并且不与任何端点关联。为了建立消息流，套接字必须先使用`zmq_connect(3)`连接到至少一个端点，或者必须创建至少一个端点来接受使用`zmq_bind(3)`的传入连接。  
>         ![在这里插入图片描述][20210310163313507.png]

> 与传统套接字的主要区别

*  一般来说，传统套接字为面向连接的字节流(SOCK\_STREAM)或不可靠数据报(SOCK\_DGRAM)提供同步接口
 *  相比之下，ZMQ套接字提供了**异步消息队列**的抽象，其确切的队列语义取决于所使用的套接字类型
 *  在传统套接字传输字节流或离散数据报流的地方，ØMQ套接字传输离散消息。
 *  ØMQ套接字是异步的，这意味着物理连接的建立，拆除，重新连接和有效交付的时间对用户是透明的，并由ØMQ自己组织。此外，在对等端不可接收消息的情况下，消息可以排队。
 *  传统套接字只允许严格的一对一(两个对等体)、多对一(多个客户端、一个服务器)或在某些情况下一对多(多播)关系。除了ZMQ\_PAIR之外，ØMQ套接字可以使用zmq\_connect()连接到多个端点，同时使用zmq\_bind()接受来自绑定到套接字的多个端点的传入连接，从而允许多对多关系。

> 线程安全

*  ØMQ既有线程安全的套接字类型，也有线程安全的套接字类型。应用程序不能在多个线程中使用非线程安全套接字，除非将套接字从一个线程迁移到另一个具有“完全隔离”内存屏障的线程。
 *  下面是线程安全套接字:`* ZMQ_CLIENT * ZMQ_SERVER * ZMQ_DISH * ZMQ_RADIO * ZMQ_SCATTER * ZMQ_GATHER`

> **type类型：**

**客户端-服务器模式**：

*  客户机-服务器模式用于允许单个ZMQ\_SERVER服务器与一个或多个ZMQ\_CLIENT客户机通信。
 *  客户端总是启动对话，在此之后任何一个对等端都可以异步地向另一个发送消息。
 *  客户机-服务器模式由http://rfc.zeromq.org/spec:41正式定义。

>  *  **ZMQ\_CLIENT**
>  *  **ZMQ\_SERVER**
>     
>      *  一个ZMQ\_SERVER套接字与一组ZMQ\_CLIENT套接字对话。

**无线电天线模式：**

*  无线电天线模式用于以扇出方式将数据从单个发布者分发到多个订阅者。
 *  Radio-dish使用组。dish套接字可以加入一个组，并且由Radio套接字发送的每条消息都属于一个组。
 *  组是以null结尾的字符串，长度限制为16个字符（包括null）。其目的是将长度增加到40个字符（包括空字符）。分组编码应为UTF8。
 *  使用精确匹配(PubSub使用前缀匹配)来匹配组。

>  *  **ZMQ\_RADIO**
>  *  **ZMQ\_DISH**

**发布-订阅模式：**

*  发布-订阅模式用于以扇出方式将数据从单个发布者一对多分发到多个订阅者。
 *  发布-订阅模式由http：// rfc.zeromq.org/spec:29正式定义。

>  *  **ZMQ\_PUB**
>     
>      *  发布者使用ZMQ\_PUB类型的套接字来分发数据。发送的消息以扇出的方式分发到所有连接的对等体。这个套接字类型没有实现`zmq_recv(3)`函数。
>  *  **ZMQ\_SUB**
>  *  **ZMQ\_XPUB**
>  *  **ZMQ\_XSUB**
>     
>      *  与ZMQ\_SUB相同，不同之处是通过向套接字发送订阅消息来进行订阅。
>      *  订阅消息是字节1（用于订阅）或字节0（用于取消订阅），后跟订阅主体。
>      *  也可以发送不带sub / unsub前缀的消息，但不会影响订阅状态。  
>         ![在这里插入图片描述][2021031016210026.png]

**流水线模式**

*  流水线模式用于将数据分发到布置在流水线中的节点。
 *  数据始终沿流水线向下流动，流水线的每一级都连接到至少一个节点。
 *  当流水线级连接到多个节点时，数据在所有连接的节点之间进行轮询。
 *  流水线模式由http：// rfc.zeromq.org/spec:30正式定义。

>  *  **ZMQ\_PUSH**
>  *  **ZMQ\_PULL**
>     
>      *  管道节点使用ZMQ\_PULL类型的套接字来接收来自上游管道节点的消息。
>      *  来自所有连接的上游节点的消息公平排队。
>      *  这个套接字类型没有实现zmq\_send()函数。

**独家配对模式**

*  exclusive pair模式用于将一个对等点精确地连接到另一个对等点。
 *  此模式用于跨inproc传输的线程间通信。
 *  独家配对模式由http://rfc.zeromq.org/spec:31正式定义。

>  *  **ZMQ\_PAIR**

**本地模式**

*  本机模式用于与TCP对等方通信，并允许在任一方向上进行异步请求和答复。

>  *  **ZMQ\_STREAM**

**请求-答复模式**

*  请求-答复模式用于将请求从ZMQ\_REQ客户端发送到一个或多个ZMQ\_REP服务，并接收对发送的每个请求的后续答复。
 *  请求/回复模式由http：// rfc.zeromq.org/spec:28正式定义。

>  *  **ZMQ\_REQ**
>     
>      *  客户端使用ZMQ\_REQ类型的套接字向服务发送请求并从服务接收答复。
>      *  此套接字类型仅允许zmq\_send（request）和后续zmq\_recv（reply）调用的交替序列。
>      *  发送的每个请求都在所有服务中轮流轮询，并且接收到的每个答复都与上一个发出的请求匹配。
>      *  如果没有服务可用，则套接字上的任何发送操作都应阻塞，直到至少一项服务可用为止。REQ套接字不得丢弃消息。  
>         ![在这里插入图片描述][20210310163508335.png]
>  *  **ZMQ\_REP**
>     
>      *  服务使用ZMQ\_REP类型的套接字来接收来自客户端的请求并向客户端发送回复。
>      *  此套接字类型仅允许zmq\_recv（request）和后续zmq\_send（reply）调用的交替序列。
>      *  接收到的每个请求都是来自所有客户机的公平队列，发送的每个应答都被路由到发出最后一个请求的客户机。
>      *  如果原始请求者不再存在，则答复将被静默丢弃。  
>         ![在这里插入图片描述][20210310163557504.png]
>  *  **ZMQ\_DEALER**
>  *  **ZMQ\_ROUTER**

## zmq\_setsockopt ##

>  *  **返回值**：
>     
>      *  如果成功，zmq\_setsockopt（）函数应返回零。否则，它将返回-1并将errno设置为以下定义的值之一。

>  *  **ZMQ\_ROUTER\_NOTIFY**
>  *  **作用**：
>     
>      *  发送连接和断开连接通知
>      *  在ROUTER套接字上启用连接和断开连接通知。启用后，当对等方连接或断开连接时，套接字将传递零长度的消息（以routing-id作为第一帧）。可以通过对标志值进行“或”操作来通知对等方的两个事件。此选项仅适用于面向流（tcp，ipc）的传输。
>      *  处于草稿状态，尚未在稳定版本中提供。

[C_C_ZeroMQ_windows_centos]: https://blog.csdn.net/zhizhengguan/article/details/114626114
[http_api.zeromq.org_master_zmq-version]: http://api.zeromq.org/master:zmq-version
[2021031014244513.png]: /images/20221023/1f0637a8950c4abfb80ffc64eb148d76.png
[http_api.zeromq.org_master_zmq-ctx-new]: http://api.zeromq.org/master:zmq-ctx-new
[http_api.zeromq.org_master_zmq-ctx-set]: http://api.zeromq.org/master:zmq-ctx-set
[http_api.zeromq.org_master_zmq-ctx-get]: http://api.zeromq.org/master:zmq-ctx-get
[http_api.zeromq.org_master_zmq-ctx-term]: http://api.zeromq.org/master:zmq-ctx-term
[http_api.zeromq.org_master_zmq-socket]: http://api.zeromq.org/master:zmq-socket#toc22
[20210310163313507.png]: /images/20221023/07ebdfd24d1a463788968b80ce215a40.png
[2021031016210026.png]: /images/20221023/4f60a2fe09104a27bfb25bf754e9d171.png
[20210310163508335.png]: /images/20221023/06eea6dffda74926965043cac06c0113.png
[20210310163557504.png]: /images/20221023/deea1d27d35442418cbed81d38b62d4a.png