如何在ZeroMQ中使用单个PULL套接字连接多个端口？

我有一个项目，多个基于Node.js实现的ZMQ PUSH客户端向不同TCP端口推送数据，需要通过一个C++应用接收这些数据。目前我采用不同线程分别从每个端口PULL数据的方案，请问ZeroMQ是否提供内置方法，支持单个套接字连接多个端口？

当前拉取线程实现（每个线程连接不同端口） #

uint16_t port = 4001;
void *context_pull = zmq_ctx_new();
void *requester_pull = zmq_socket(context_pull, ZMQ_PULL);
zmq_setsockopt(requester_pull, ZMQ_RCVTIMEO, &timeout, sizeof(int));

int len = sprintf(pub_url, "%s%s%u", (char *) "tcp://", "127.0.0.1:", port);
pub_url[len] = '\0';

int pull = zmq_connect(requester_pull, pub_url);
if (pull < 0) {
    printf("CAN NOT BIND TO PORT %s\n", pub_url);
} else {
    printf("PULL PROCESS CONNECTED TO PORT %s\n", pub_url);
}

char buf[256];

while (1) {
    int rep = zmq_recv(requester_pull, buf, sizeof (buf), 0);
    if (rep > 0) {
        add_log(json::parse(std::string(buf)));  // 此方法处理接收到的JSON数据
    } else {
        printf("DATA PULL ERROR!\n");
    }
}

ZeroMQ完全支持单个ZMQ_PULL套接字连接多个TCP端口，无需为每个端口单独创建线程和套接字，能大幅简化你的代码结构。

核心实现方式 #

对同一个ZMQ_PULL套接字多次调用zmq_connect()，每次传入不同的目标端口地址即可。ZeroMQ会自动管理所有连接，当任意连接有数据到达时，zmq_recv()都能获取到数据，默认采用*公平队列（fair-queuing）*策略轮询从各个连接接收数据，避免单个连接的数据被饥饿。

修改后的示例代码 #

void *context_pull = zmq_ctx_new();
void *requester_pull = zmq_socket(context_pull, ZMQ_PULL);
zmq_setsockopt(requester_pull, ZMQ_RCVTIMEO, &timeout, sizeof(int));

// 连接多个目标端口
uint16_t ports[] = {4001, 4002, 4003};
char pub_url[64];

for (size_t i = 0; i < sizeof(ports)/sizeof(ports[0]); i++) {
    int len = sprintf(pub_url, "tcp://127.0.0.1:%u", ports[i]);
    pub_url[len] = '\0';
    
    int pull = zmq_connect(requester_pull, pub_url);
    if (pull < 0) {
        printf("无法连接到端口 %s\n", pub_url);
    } else {
        printf("PULL 已连接到端口 %s\n", pub_url);
    }
}

char buf[256];

while (1) {
    int rep = zmq_recv(requester_pull, buf, sizeof(buf), 0);
    if (rep > 0) {
        add_log(json::parse(std::string(buf)));  // 处理接收到的JSON数据
    } else {
        printf("数据拉取错误！\n");
    }
}

关键注意点 #

• 单个ZMQ_PULL套接字的多连接是线程安全的，但如果要在多个线程中调用zmq_recv()，需要额外加锁或启用ZeroMQ的线程安全套接字选项；通常单线程处理所有接收逻辑更简单高效。
• 后续若需动态添加新连接，直接在主线程中再次调用zmq_connect()即可，无需重启套接字。

内容的提问来源于stack exchange，提问作者J.Maduka