ZeroMQ DEALER连接多ROUTER时断连节点的故障转移问题

你遇到的这个问题确实是ZeroMQ DEALER套接字的默认行为——它会持续维护所有已建立的连接端点，哪怕其中一个ROUTER意外下线，DEALER依然会按轮询策略尝试往这个失效端点发送消息，这些消息会被存在本地队列中（直到队列达到高水位或端点恢复），导致正常的ROUTER只能收到一半流量。

好消息是，ZeroMQ确实内置了检测死连接并自动跳过失效端点的机制，不需要你手动实现复杂的心跳、ACK或单独套接字逻辑，只需要配置几个关键参数即可。

一、启用ZeroMQ心跳机制 #

ZeroMQ的心跳参数可以让套接字定期发送心跳包，自动检测连接是否存活。当连接超时未响应时，ZeroMQ会将该端点标记为不可用，DEALER就不会再往这个端点发送消息了。

你需要在DEALER套接字上设置以下参数：

• ZMQ_HEARTBEAT_IVL：心跳包发送间隔（毫秒），比如设为1000ms，每秒发一次心跳
• ZMQ_HEARTBEAT_TIMEOUT：等待心跳响应的超时时间（毫秒），比如3000ms，超过3秒没收到响应就判定连接失效
• ZMQ_HEARTBEAT_TTL：心跳包的存活时间（毫秒），比如2000ms，确保心跳包不会在网络中滞留过久

在你的Go代码中，创建DEALER套接字后添加这些配置：

dealer, _ := zmq4.NewSocket(zmq4.DEALER)
// 配置心跳参数
dealer.SetHeartbeatIvl(1000)
dealer.SetHeartbeatTimeout(3000)
dealer.SetHeartbeatTtl(2000)

二、配合TCP保活机制（可选） #

如果你的传输协议是TCP，还可以开启TCP层的保活机制，和ZeroMQ心跳形成双重保障：

// 开启TCP保活
dealer.SetTcpKeepalive(1)
// 连接空闲3秒后开始发送保活包
dealer.SetTcpKeepaliveIdle(3000)
// 保活包发送间隔1秒
dealer.SetTcpKeepaliveIntvl(1000)
// 连续3次保活包无响应则判定连接断开
dealer.SetTcpKeepaliveCnt(3)

三、优化消息队列行为 #

为了避免失效端点的消息一直占用本地队列，你可以设置ZMQ_LINGER参数为0，这样当连接被判定为失效时，队列中未发送的消息会被立即丢弃，而不是一直等待：

dealer.SetLinger(0)

修改后的完整测试代码 #

把这些配置整合到你的测试代码中，修改后的main函数如下：

func main() {
    dealer, _ := zmq4.NewSocket(zmq4.DEALER)
    router1, _ := zmq4.NewSocket(zmq4.ROUTER)
    router2, _ := zmq4.NewSocket(zmq4.ROUTER)

    // 配置DEALER的心跳、TCP保活和linger参数
    dealer.SetHeartbeatIvl(1000)
    dealer.SetHeartbeatTimeout(3000)
    dealer.SetHeartbeatTtl(2000)
    dealer.SetTcpKeepalive(1)
    dealer.SetTcpKeepaliveIdle(3000)
    dealer.SetTcpKeepaliveIntvl(1000)
    dealer.SetTcpKeepaliveCnt(3)
    dealer.SetLinger(0)

    router1.Bind("tcp://0.0.0.0:6667")
    router2.Bind("tcp://0.0.0.0:6668")

    dealer.Connect("tcp://0.0.0.0:6667")
    dealer.Connect("tcp://0.0.0.0:6668")

    router1.SetSubscribe("")
    router2.SetSubscribe("")
    dealer.SetSubscribe("")

    // 正常场景测试
    for i := 0; i < 10; i++ {
        dealer.SendBytes([]byte("Hello World"), 0)
    }
    time.Sleep(300 * time.Millisecond)
    count1 := receiveAll(router1)
    count2 := receiveAll(router2)
    fmt.Printf("Blue sky scenario: count1=%d count2=%d\n", count1, count2)

    // 关闭ROUTER1
    router1.Close()
    // 等待心跳检测到连接失效
    time.Sleep(4000 * time.Millisecond) // 等待超过心跳超时时间

    // 故障场景测试
    for i := 0; i < 10; i++ {
        dealer.SendBytes([]byte("Hello World"), 0)
    }
    time.Sleep(300 * time.Millisecond)
    count := receiveAll(router2)
    fmt.Printf("Peer 1 offline: count=%d\n", count)
}

效果说明 #

修改后，当ROUTER1下线，DEALER会在心跳超时（3秒）后检测到连接失效，后续的10条消息会全部发送到ROUTER2，最终输出Peer 1 offline: count=10，符合你的需求。

需要注意的是，要给足够的时间让ZeroMQ完成连接失效检测（比如代码中等待4秒），确保DEALER已经把失效端点从可用列表中移除。

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