ZeroMQ IO多路复用时轮询逻辑 zmq_poll

62 阅读 0 评论 41 点赞

我是靠谱客的博主爱笑柜子，最近开发中收集的这篇文章主要介绍ZeroMQ IO多路复用时轮询逻辑 zmq_poll，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

背景简介

一个zmq的服务端可以多次调用zmq_bind()以将自己关联到多个endpoint上.这就意味着, zmq_socket可以同时接受来自多个不同通讯协议的多簇请求消息.同理，一个zmq客户端可以多次调用zmq_connect()连接到多个server上。

zmq_bind(socket, "tcp://*:5555"); 
zmq_bind(socket, "tcp://*:999");

zmq_connect(socket, "tcp://*:5555"); 
zmq_connect(socket, "tcp://*:999");

但问题是如何优雅的做多个端口的事件监听，查询每个端口是否有消息传进来呢？

解决方法

zmq_poll方法可以查询每个端口是否有事件进来。此方法需要三个参数：

items_: 是一个以 zmq_pollitem_t 为数据类型的数组，包含着要轮询的对象。
nitems：是要轮询的对象个数，也就是items的数组大小
timeout_：等待时间毫秒计算，如果设置为大于零的整数，意味着等待时间，超时后则返回。设置为等于零，则不会进行等待，判断items后直接返回。设置为-1时会永久等待，直到items有事件返回。
```
   int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_);
```
zmq_poll 返回值是整数，代表有事件进来的数量。

zmq_pollitem_t数据结构

typedef struct  
{  
    void *socket;  //要轮询的zmq套接字
    int   fd;      //句柄 （file descrption）一般我们定义socket即可，不会使用fd
    short events;  //要轮询的事件  ex. ZMQ_POLLIN： 套接字接收消息事件
    short revents; //轮询后返回的事件，若有事件发生，zmq_poll方法会进行设置 
} zmq_pollitem_t;

简单案例

代码能够同时监听8000和8002端口发送过来的数据

#include "include/zmq.h"
#include <iostream>

int main()
{
   void *context = zmq_ctx_new();
   void *proxy_first = zmq_socket(context, ZMQ_REP);
   void *proxy_second = zmq_socket(context, ZMQ_REP);

   zmq_bind(proxy_first, "tcp://*:8000");
   zmq_bind(proxy_second, "tcp://*:8002");
   
   //创建2个 zmq_pollitem_t 类型的数据，并放入items数组里。
   //第一个zmq_pollitem_t类型数据 socket： proxy_first； fd：0； events：ZMQ_POLLIN； revents：0
   //第二个zmq_pollitem_t类型数据 socket： proxy_second； fd：0； events：ZMQ_POLLIN； revents：0
   zmq_pollitem_t items[] = {
       {proxy_first, 0, ZMQ_POLLIN, 0},
       {proxy_second, 0, ZMQ_POLLIN, 0},
   };
   
   //进入无线循环，不断进行轮询
   while (1)
   {
       zmq_msg_t message;
       //将items里的两个zmq_pollitem_t进行轮询，并一直阻塞到有POLLIN事件出现后
       zmq_poll(items, 2, -1);
       
       // 检察第一个item的返回事件是否是ZMQ_POLLIN，如果是则说明第一个item有ZMQ_POLLIN返回事件
       if (items[0].revents & ZMQ_POLLIN)
       {
           //循环打印接收到的信息，直到信息最后
           while (1)
           {
               zmq_msg_init(&message);
               zmq_msg_recv(&message, proxy_first, 0);
               int more = zmq_msg_more(&message);
               //将收到的信息打印出来
               std::cout<<message<<std::endl;
               zmq_msg_close(&message);
               if (!more)
               {
                   break;
               }
           }
       }
       // 检察第二个item的返回事件是否是ZMQ_POLLIN
       if (items[1].revents & ZMQ_POLLIN)
       {
           while (1)
           {
               zmq_msg_init(&message);
               zmq_msg_recv(&message, proxy_second, 0);
               int more = zmq_msg_more(&message);
               //将收到的信息打印出来
               std::cout<<message<<std::endl;
               zmq_msg_close(&message);
               if (!more)
               {
                   break;
               }
           }
       }
   }
}