一、ZeroMQ高级模式：路由-代理-经销商

ZeroMQ 的内置模式里，我最常用也最头疼的就是这套 ROUTER-DEALER 组合。说白了，它解决了一个核心问题：如何让多个客户端和多个服务端高效通信。

1.1 ROUTER 套接字

ROUTER 是个异步套接字。它不像 REQ 那样必须一问一答。每个消息进来时，ROUTER 会自动带上一个身份帧。这个帧就像快递单号，标记了消息从哪来。

关键点：ROUTER 可以同时接收多个客户端的请求，并且能分别回复。它不会搞混消息。

// ROUTER 端代码示例
void *context = zmq_ctx_new();
void *router = zmq_socket(context, ZMQ_ROUTER);
zmq_bind(router, "tcp://*:5555");

while (1) {
    zmq_msg_t identity;
    zmq_msg_t msg;
    zmq_msg_init(&identity);
    zmq_msg_init(&msg);
    
    // 先收身份帧
    zmq_msg_recv(&identity, router, 0);
    // 再收数据帧
    zmq_msg_recv(&msg, router, 0);
    
    printf("收到来自 [%s] 的消息: %s\n", 
           (char*)zmq_msg_data(&identity),
           (char*)zmq_msg_data(&msg));
    
    // 回复时，先发身份帧，再发数据
    zmq_msg_send(&identity, router, ZMQ_SNDMORE);
    zmq_msg_send(&msg, router, 0);
    
    zmq_msg_close(&identity);
    zmq_msg_close(&msg);
}

1.2 DEALER 套接字

DEALER 是 ROUTER 的对称搭档。它也是个异步套接字。DEALER 发送消息时，可以不带身份帧——它会自动分配一个随机身份。

我个人习惯把 DEALER 用在客户端。为什么？因为它可以同时发送多个请求，不用等回复。这在做批量任务时特别爽。

避坑指南：我曾经在项目中直接用 REQ 套接字做批量请求，结果卡死在 send/recv 的严格交替上。换成 DEALER 后，问题立刻解决。

1.3 路由-代理-经销商模式

这个模式的结构是这样的：

ROUTER-DEALER 代理模式 客户端 1 (DEALER) 客户端 2 (DEALER) 客户端 3 (DEALER) 代理 (Proxy) 前端: ROUTER 后端: DEALER 服务端 1 (ROUTER) 服务端 2 (ROUTER) 服务端 3 (ROUTER) 客户端通过 DEALER 发送请求到代理 代理用 ROUTER 接收，用 DEALER 转发给服务端 服务端用 ROUTER 接收并回复

这个模式的核心是代理层。代理内部维护一个 ROUTER 和一个 DEALER。ROUTER 面向客户端，DEALER 面向服务端。代理负责转发消息，同时可以做负载均衡、消息过滤等操作。

二、ZeroMQ的可靠性设计

ZeroMQ 本身不保证消息可靠送达。它只保证尽力交付。但实际项目中，消息丢了可是大事。我来说说怎么搞可靠性。

2.1 心跳机制

最简单的可靠性手段就是心跳。客户端定期发一个心跳包给服务端。如果服务端一段时间没收到心跳，就认为客户端挂了。

// 心跳发送端（客户端）
void *heartbeat_sender(void *arg) {
    void *socket = (void*)arg;
    while (1) {
        zmq_send(socket, "HB", 2, 0);
        sleep(1);  // 每秒发一次
    }
    return NULL;
}

// 心跳检测端（服务端）
void check_heartbeat(time_t last_hb, int timeout) {
    time_t now = time(NULL);
    if (difftime(now, last_hb) > timeout) {
        printf("客户端心跳超时，可能已断开\n");
        // 执行清理逻辑
    }
}

2.2 消息确认机制

我曾经在一个金融交易系统里吃过亏。消息发出去后，服务端处理完了，但回复在半路丢了。客户端以为没处理，又发了一遍——结果重复交易了。

解决方案：每个消息带唯一ID，服务端处理完后回复ACK。客户端收到ACK才认为消息已处理。

注意：消息去重是可靠性设计中最容易被忽视的一环。一定要在应用层做幂等处理。

2.3 重连机制

ZeroMQ 有自动重连功能。但默认参数不一定适合你的场景。我建议手动配置重连间隔和重试次数。

// 设置重连间隔为100ms
int reconnect_ivl = 100;
zmq_setsockopt(socket, ZMQ_RECONNECT_IVL, &reconnect_ivl, sizeof(reconnect_ivl));

// 设置最大重连间隔为3000ms
int reconnect_ivl_max = 3000;
zmq_setsockopt(socket, ZMQ_RECONNECT_IVL_MAX, &reconnect_ivl_max, sizeof(reconnect_ivl_max));

三、ZeroMQ的代理与转发

代理在 ZeroMQ 里是个很实用的东西。它不处理业务逻辑，只负责转发消息。但别小看它，代理能解决很多架构问题。

3.1 内置代理函数

ZeroMQ 提供了一个 zmq_proxy() 函数。一行代码就能启动代理：

void *frontend = zmq_socket(context, ZMQ_ROUTER);
void *backend = zmq_socket(context, ZMQ_DEALER);
zmq_bind(frontend, "tcp://*:5555");
zmq_bind(backend, "tcp://*:5556");

// 启动代理，阻塞运行
zmq_proxy(frontend, backend, NULL);

这个函数内部做了所有转发逻辑。你想想看，要是自己写，得处理多少边界情况？

3.2 自定义代理

但有时候内置代理不够用。比如你想在转发时记录日志，或者做流量控制。这时候就得自己写代理循环。

void custom_proxy(void *frontend, void *backend) {
    zmq_pollitem_t items[] = {
        {frontend, 0, ZMQ_POLLIN, 0},
        {backend, 0, ZMQ_POLLIN, 0}
    };
    
    while (1) {
        zmq_poll(items, 2, -1);
        
        if (items[0].revents & ZMQ_POLLIN) {
            // 从客户端收到消息，转发给服务端
            zmq_msg_t msg;
            zmq_msg_init(&msg);
            zmq_msg_recv(&msg, frontend, 0);
            // 这里可以加日志
            printf("转发客户端消息\n");
            zmq_msg_send(&msg, backend, 0);
            zmq_msg_close(&msg);
        }
        
        if (items[1].revents & ZMQ_POLLIN) {
            // 从服务端收到消息，转发给客户端
            zmq_msg_t msg;
            zmq_msg_init(&msg);
            zmq_msg_recv(&msg, backend, 0);
            zmq_msg_send(&msg, frontend, 0);
            zmq_msg_close(&msg);
        }
    }
}

四、ZeroMQ的安全机制

安全这块，ZeroMQ 提供了 CURVE 加密。它基于椭圆曲线加密，性能不错。

4.1 CURVE 加密配置

配置 CURVE 需要三步：生成密钥对、服务端设置密钥、客户端设置密钥。

// 服务端
zmq_curve_keypair(server_public, server_secret);
zmq_setsockopt(socket, ZMQ_CURVE_SERVER, &yes, sizeof(yes));
zmq_setsockopt(socket, ZMQ_CURVE_SECRETKEY, server_secret, 40);

// 客户端
zmq_setsockopt(socket, ZMQ_CURVE_SERVERKEY, server_public, 40);
zmq_setsockopt(socket, ZMQ_CURVE_PUBLICKEY, client_public, 40);
zmq_setsockopt(socket, ZMQ_CURVE_SECRETKEY, client_secret, 40);

避坑指南：我曾经把公钥和密钥搞反了，结果调试了一整天。记住：服务端只需要设置自己的密钥，客户端需要设置服务端的公钥。

4.2 安全最佳实践

  • 密钥定期轮换：不要一个密钥用一年
  • 使用TLS隧道：如果CURVE不够用，可以配合TLS
  • 白名单机制：在代理层做IP过滤

五、ZeroMQ的性能调优

性能调优是门手艺活。我踩过不少坑，总结几个关键点。

5.1 缓冲区大小

默认的缓冲区可能不够。特别是高吞吐场景，缓冲区满了就会丢消息。

// 设置发送缓冲区为1MB
int sndbuf = 1024 * 1024;
zmq_setsockopt(socket, ZMQ_SNDBUF, &sndbuf, sizeof(sndbuf));

// 设置接收缓冲区为1MB
int rcvbuf = 1024 * 1024;
zmq_setsockopt(socket, ZMQ_RCVBUF, &rcvbuf, sizeof(rcvbuf));

5.2 消息大小

ZeroMQ 对小于 1KB 的消息处理最快。超过 1MB 的消息，性能会急剧下降。我建议：大消息分片，或者用文件传输代替。

5.3 多线程优化

ZeroMQ 的上下文可以跨线程使用。但每个套接字只能在一个线程里操作。我习惯用 zmq_poll 做事件驱动，避免多线程锁竞争。

参数 默认值 推荐值 说明
ZMQ_SNDBUF 0 (系统默认) 1MB 发送缓冲区
ZMQ_RCVBUF 0 (系统默认) 1MB 接收缓冲区
ZMQ_RECONNECT_IVL 100ms 100ms 重连间隔
ZMQ_LINGER -1 (无限等待) 0 关闭时丢弃未发送消息

六、ZeroMQ的应用场景

ZeroMQ 适合哪些场景？我列几个实际用过的：

  • 微服务通信：服务间轻量级消息传递
  • 实时数据流：股票行情、物联网数据采集
  • 任务分发：把大任务拆成小任务，分发给多个worker
  • 日志收集：多个服务把日志发到中心收集器

不适合的场景：需要严格消息顺序的（可以用Kafka）、需要持久化存储的（可以用RabbitMQ）。

七、ZeroMQ综合案例

最后来个综合案例：一个简单的分布式任务处理系统。

// 任务分发器（代理）
void *frontend = zmq_socket(context, ZMQ_ROUTER);
void *backend = zmq_socket(context, ZMQ_DEALER);
zmq_bind(frontend, "tcp://*:5555");
zmq_bind(backend, "tcp://*:5556");

// 启动代理
zmq_proxy(frontend, backend, NULL);

// Worker 端
void *worker = zmq_socket(context, ZMQ_DEALER);
zmq_connect(worker, "tcp://localhost:5556");

while (1) {
    zmq_msg_t msg;
    zmq_msg_init(&msg);
    zmq_msg_recv(&msg, worker, 0);
    
    // 处理任务
    char *task = (char*)zmq_msg_data(&msg);
    printf("Worker 处理任务: %s\n", task);
    
    // 发送结果
    zmq_send(worker, "DONE", 4, 0);
    zmq_msg_close(&msg);
}

这个案例里，代理负责分发任务，Worker 处理完后回复结果。你想想看，如果不用 ZeroMQ，自己实现这套东西得写多少代码？

总结：ZeroMQ 的高级模式、可靠性设计、安全机制和性能调优，都是实际项目中必须掌握的技能。别光看文档，动手写代码才是最快的成长方式。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →