c++如何使用zeromq消息队列_c++ 发布订阅模式与请求响应模式【实战】

冰火之心

发布时间：2026-01-01 13:13:02

762人浏览过

来源于php中文网

原创

ZeroMQ的PUB/SUB与REQ/REP是两种根本不同的通信模式：前者单向广播、无连接、需显式订阅；后者双向配对、强顺序、send/recv必须严格交替，socket类型不可混用或复用。

c++如何使用zeromq消息队列_c++ 发布订阅模式与请求响应模式【实战】

ZeroMQ 的 C++ 发布订阅（PUB/SUB）和请求响应（REQ/REP）模式，本质是两种完全不同的通信契约：前者是单向广播、无连接、无确认；后者是双向配对、隐式同步、强顺序。选错模式或混用语义，90% 的“收不到消息”“卡死”“乱序”问题都源于此。

`zmq::socket_t` 类型必须严格匹配模式，且不能复用

ZeroMQ 不允许一个 socket 同时承担多种角色。比如用 ZMQ_REQ socket 去 connect 一个 ZMQ_PUB 端口，程序不会报错，但永远收不到任何数据——因为协议层根本不兼容。

ZMQ_REQ 只能 connect 到 ZMQ_REP（或 ZMQ_ROUTER），且每次 send 后必须紧跟 recv，否则下一次 send 会阻塞
ZMQ_PUB 只能 bind（不建议 connect），ZMQ_SUB 只能 connect（不建议 bind），且 ZMQ_SUB 必须调用 setsockopt(ZMQ_SUBSCRIBE, ...) 才能收到消息，空订阅（""）在较新版本中默认不生效
一个 zmq::context_t 可创建多个 socket，但每个 socket 的类型和行为由构造时第二个参数唯一决定，不可运行时变更

发布订阅模式中，`zmq_setsockopt(..., ZMQ_SUBSCRIBE, ...)` 是必填项

这是新手最常踩的坑：SUB 端代码写了 connect，也写了 recv，但始终收不到 PUB 发出的消息。根本原因就是漏了订阅动作——ZeroMQ 的 SUB socket 默认屏蔽所有消息，哪怕 PUB 已经在发。

订阅空字符串：subscriber.setsockopt(ZMQ_SUBSCRIBE, "", 0); 表示接收所有主题（注意长度传 0，不是 1）
订阅特定前缀：std::string topic = "stock.AAPL"; subscriber.setsockopt(ZMQ_SUBSCRIBE, topic.c_str(), topic.size());
可多次调用 setsockopt(ZMQ_SUBSCRIBE, ...) 实现多主题订阅，但不支持通配符（如 "stock.*"）——需靠应用层过滤
取消订阅用 setsockopt(ZMQ_UNSUBSCRIBE, ...)，但实际中极少需要

请求响应模式里，`send` 和 `recv` 必须严格交替，且不能跳过响应

ZMQ_REQ socket 内部维护一个“请求-响应”状态机。一旦发出请求未收到回复，socket 就会进入不可用状态，后续 send 直接返回 false 或抛异常（取决于 cppzmq 版本）。

XPaper Ai

AI撰写论文、开题报告生成、AI论文生成器尽在XPaper Ai论文写作辅助指导平台

下载

立即学习“C++免费学习笔记（深入）”；

服务端 ZMQ_REP 也必须先 recv 再 send，否则客户端会一直等待
不要试图在一个 REQ socket 上并发发多个请求——它不支持异步 I/O；如需并发，请用多个 socket 或改用 ZMQ_DEALER + ZMQ_ROUTER
超时控制推荐用 setsockopt(ZMQ_RCVTIMEO, &timeout_ms, sizeof(timeout_ms))，避免无限阻塞；但注意：ZMQ_SNDTIMEO 对 REQ socket 无效

#include 
#include 
#include 

// 客户端：REQ 模式（务必配对 recv）
int main() {
    zmq::context_t ctx;
    zmq::socket_t sock(ctx, ZMQ_REQ);
    sock.connect("tcp://localhost:5555");

    // 设置接收超时，防止永久卡住
    int timeout_ms = 3000;
    sock.setsockopt(ZMQ_RCVTIMEO, &timeout_ms, sizeof(timeout_ms));

    zmq::message_t req(5);
    memcpy(req.data(), "HELLO", 5);
    sock.send(req);

    zmq::message_t reply;
    if (sock.recv(reply) == 0) {  // 返回 0 表示成功
        std::cout << "Got: " << std::string(static_cast(reply.data()), reply.size()) << "\n";
    } else {
        std::cerr << "Timeout or error\n";
    }
}