ZMQ_ROUTER语义疑问：未丢弃断开连接客户端的消息

我尝试用ZMQ_REQ和ZMQ_ROUTER实现可靠的多客户端单服务器req/rep模式，根据ZeroMQ RFC，ROUTER在对等端断开时会销毁双队列并丢弃其中所有消息。但测试中发现，服务端模拟CPU过载休眠20秒后，仍会处理之前断开客户端的消息，不符合预期。

测试代码 #

客户端代码（cppzmq v4.10.0，libzmq v4.3.5）

#include <zmq.hpp>
#include <iostream>

zmq::context_t context{1};

bool reqrep(zmq::message_t &req, std::string addr)
{
    zmq::socket_t client{context, zmq::socket_type::req};
    client.set(zmq::sockopt::rcvtimeo, 2500);
    client.set(zmq::sockopt::sndtimeo, 2500);
    client.set(zmq::sockopt::immediate, true);
    client.set(zmq::sockopt::linger, 0);

    client.connect(addr);
    if (!client.send(req))
    {
        std::cerr << "E: send timeout\n";
        return false;
    }

    zmq::message_t rep;
    if (client.recv(rep))
    {
        std::cout << "I: server replied OK (" << rep.to_string() << ")\n";
        return true;
    }
    else
    {
        std::cerr << "E: receive timeout\n";
        return false;
    }
}

int main()
{
    std::string addr{"ipc:///tmp/server"};

    int seq{};
    while (true)
    {
        zmq::message_t req{std::to_string(seq++)};
        reqrep(req, addr);
    }

    return 0;
}

服务端代码（cppzmq v4.10.0，libzmq v4.3.5）

#include <unistd.h>
#include <zmq_addon.hpp>
#include <iostream>

//  Provide random number from 0..(num-1)
#define within(num) (int)((float)((num) * random()) / (RAND_MAX + 1.0))

int main()
{
    zmq::context_t context(1);
    zmq::socket_t server(context, ZMQ_ROUTER);

    server.bind("ipc:///tmp/server");

    while (1)
    {
        zmq::multipart_t mp;
        auto req = zmq::recv_multipart(server, std::back_inserter(mp));
        std::cout << "I: message received";
        std::cout << mp.str() << std::endl;
        if (!within(20))
        {
            std::cout << "I: simulating CPU overload" << std::endl;
            sleep(20);
        }
        sleep(1); // Do work
        zmq::send_multipart(server, mp);
    }
    return 0;
}

问题原因分析 #

• ROUTER队列语义的范围：RFC中提到的“销毁双队列并丢弃消息”，仅针对ROUTER套接字内部未被应用层接收的消息。一旦服务端通过recv_multipart将消息读取到应用层内存（比如你的zmq::multipart_t mp），这些消息就脱离了ROUTER的管理，后续客户端断开也不会影响服务端对它们的处理。
• 客户端连接生命周期不匹配：你的客户端每次请求都创建新的REQ套接字，发送消息后等待2.5秒超时就关闭。但服务端调用recv_multipart时，客户端可能仍处于连接状态；当服务端休眠20秒期间客户端才断开，此时消息已经在服务端应用层，ROUTER无法干预。
• 断开检测的延迟：ZeroMQ默认依赖TCP的默认Keepalive机制（通常为分钟级），不会即时检测客户端断开。这意味着ROUTER可能在服务端已经接收消息后，才发现客户端断开，此时消息已进入处理流程。
• 服务端无状态处理逻辑：服务端接收消息后直接处理，没有在回复前检查客户端是否在线。即使ROUTER在回复时发现客户端不存在，也只是无法发送回复，但消息的处理过程已经完成。

解决方案 #

• 在回复阶段检测客户端状态：给ROUTER设置ZMQ_ROUTER_MANDATORY选项，发送回复时如果客户端已断开，zmq_send_multipart会返回错误，此时可选择丢弃处理结果：

  server.set(zmq::sockopt::router_mandatory, true);
  // 处理消息后
  auto send_result = zmq::send_multipart(server, mp, zmq::send_flags::dontwait);
  if (!send_result) {
      std::cerr << "E: client disconnected, discard reply\n";
  }
  

• 复用客户端套接字：不要每次请求都创建新的REQ套接字，复用同一个套接字可以让ROUTER更稳定地跟踪客户端连接状态，减少断开检测的延迟。
• 启用TCP Keepalive加速断开检测：给ROUTER和REQ套接字设置Keepalive选项，让系统更快检测到客户端断开：

  // 服务端ROUTER配置
  server.set(zmq::sockopt::tcp_keepalive, 1);
  server.set(zmq::sockopt::tcp_keepalive_idle, 5);    // 5秒无数据发送第一个心跳
  server.set(zmq::sockopt::tcp_keepalive_interval, 2); // 心跳间隔2秒
  server.set(zmq::sockopt::tcp_keepalive_count, 3);   // 3次心跳失败判定断开
  
  // 客户端REQ配置
  client.set(zmq::sockopt::tcp_keepalive, 1);
  // 同步配置idle、interval、count选项
  

• 跟踪客户端活跃状态：维护一个客户端ID集合，当回复失败时标记该ID为离线，后续接收到该ID的消息直接丢弃，直到收到新的有效请求再重新标记为在线。

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