ZMQ INPROC传输下PUB-SUB无法正常接收消息问题排查

问题场景 #

同一线程内搭建2组一对一的PUB-SUB通信（基于INPROC传输），订阅端read_socket_srv的zmq_recv会永久阻塞，必须先执行一次无意义的非阻塞recv操作才能正常接收后续消息。拆分进程改用IPC/TCP传输时无此问题，仅INPROC传输下出现该异常。

示例代码（原问题代码） #

#include <zmq.h>
#include <stdio.h>
#include <unistd.h>

int main() {
    void* ctx = zmq_ctx_new();

    // 第一组 PUB-SUB
    void* write_socket_cli = zmq_socket(ctx, ZMQ_PUB);
    zmq_bind(write_socket_cli, "inproc://cli_srv");
    void* read_socket_cli = zmq_socket(ctx, ZMQ_SUB);
    zmq_connect(read_socket_cli, "inproc://cli_srv");
    zmq_setsockopt(read_socket_cli, ZMQ_SUBSCRIBE, "", 0);

    // 第二组 PUB-SUB
    void* write_socket_srv = zmq_socket(ctx, ZMQ_PUB);
    zmq_bind(write_socket_srv, "inproc://srv_cli");
    void* read_socket_srv = zmq_socket(ctx, ZMQ_SUB);
    zmq_connect(read_socket_srv, "inproc://srv_cli");
    zmq_setsockopt(read_socket_srv, ZMQ_SUBSCRIBE, "", 0);

    // 问题点：注释掉下面这行，read_socket_srv的recv会永久阻塞
    char dummy[1];
    zmq_recv(read_socket_srv, dummy, sizeof(dummy), ZMQ_DONTWAIT);

    // 第一组发送消息
    const char* msg1 = "Hello from CLI";
    zmq_send(write_socket_cli, msg1, sizeof(msg1), 0);
    char buf1[256];
    zmq_recv(read_socket_cli, buf1, sizeof(buf1), 0);
    printf("CLI received: %s\n", buf1);

    // 第二组发送消息
    const char* msg2 = "Hello from SRV";
    zmq_send(write_socket_srv, msg2, sizeof(msg2), 0);
    char buf2[256];
    zmq_recv(read_socket_srv, buf2, sizeof(buf2), 0);
    printf("SRV received: %s\n", buf2);

    // 清理资源
    zmq_close(write_socket_cli);
    zmq_close(read_socket_cli);
    zmq_close(write_socket_srv);
    zmq_close(read_socket_srv);
    zmq_ctx_destroy(ctx);
    return 0;
}

原因分析 #

INPROC传输的线程内通信中，ZMQ依赖内部事件循环处理订阅者的注册逻辑。当同一线程内连续创建多组PUB-SUB时，SUB端完成ZMQ_SUBSCRIBE设置后，ZMQ并未立即同步订阅状态到PUB端的路由表。此时PUB端发送的消息会因订阅匹配未完成而被丢弃，后续SUB端的recv就会因队列无消息而永久阻塞。

而一次非阻塞recv操作会主动触发ZMQ内部的事件处理流程，强制完成订阅者的注册同步，让后续发送的消息能被正确路由到SUB端的消息队列中。

解决方案 #

替换无意义的非阻塞recv，改用zmq_poll主动等待订阅状态生效，这是更规范的处理方式：

#include <zmq.h>
#include <stdio.h>
#include <unistd.h>

int main() {
    void* ctx = zmq_ctx_new();

    // 第一组 PUB-SUB
    void* write_socket_cli = zmq_socket(ctx, ZMQ_PUB);
    zmq_bind(write_socket_cli, "inproc://cli_srv");
    void* read_socket_cli = zmq_socket(ctx, ZMQ_SUB);
    zmq_connect(read_socket_cli, "inproc://cli_srv");
    zmq_setsockopt(read_socket_cli, ZMQ_SUBSCRIBE, "", 0);

    // 第二组 PUB-SUB
    void* write_socket_srv = zmq_socket(ctx, ZMQ_PUB);
    zmq_bind(write_socket_srv, "inproc://srv_cli");
    void* read_socket_srv = zmq_socket(ctx, ZMQ_SUB);
    zmq_connect(read_socket_srv, "inproc://srv_cli");
    zmq_setsockopt(read_socket_srv, ZMQ_SUBSCRIBE, "", 0);

    // 用zmq_poll触发订阅同步，替代无意义的非阻塞recv
    zmq_pollitem_t items[] = {
        { read_socket_srv, 0, ZMQ_POLLIN, 0 }
    };
    // 等待100ms，确保订阅注册完成（INPROC下实际耗时极短）
    zmq_poll(items, 1, 100);

    // 第一组发送消息
    const char* msg1 = "Hello from CLI";
    zmq_send(write_socket_cli, msg1, sizeof(msg1), 0);
    char buf1[256];
    zmq_recv(read_socket_cli, buf1, sizeof(buf1), 0);
    printf("CLI received: %s\n", buf1);

    // 第二组发送消息
    const char* msg2 = "Hello from SRV";
    zmq_send(write_socket_srv, msg2, sizeof(msg2), 0);
    char buf2[256];
    zmq_recv(read_socket_srv, buf2, sizeof(buf2), 0);
    printf("SRV received: %s\n", buf2);

    // 清理资源
    zmq_close(write_socket_cli);
    zmq_close(read_socket_cli);
    zmq_close(write_socket_srv);
    zmq_close(read_socket_srv);
    zmq_ctx_destroy(ctx);
    return 0;
}

额外说明 #

• INPROC传输的线程内通信，ZMQ不会自动触发事件循环，必须通过主动IO操作（recv/send/poll）来驱动内部状态同步。
• PUB-SUB模式本身是异步消息分发，确保SUB端订阅生效后再发送消息，是避免消息丢失和阻塞的核心原则。

内容的提问来源于stack exchange，提问作者Francesco