ZMQ DEALER-ROUTER-DEALER模式ACK丢失致通信挂起问题求助

我搭建了一套多客户端向多服务器发送消息的ZMQ通信系统，采用DEALER->ROUTER->DEALER模式，每条消息仅指向单个服务器。客户端知晓所有可用服务器ID，仅向已连接的服务器发送消息；服务器启动后连接至套接字，多个服务器Worker绑定到inproc路由套接字，通信始终由客户端发起，消息异步发送至服务器。

当客户端和服务器Worker数量增加时，服务器发送给客户端的ACK（通信流程第8步）始终无法送达，导致客户端等待ACK、服务器等待更多消息，双方均挂起，需重启才能恢复。系统日志未发现明显异常，现寻求进一步排查的建议与指导。

配置与代码细节 #

客户端启动时以DEALER套接字连接至Broker的IP:Port：

requester, _ := zmq.NewSocket(zmq.DEALER)

Broker通过前端TCP套接字连接客户端Worker，后端inproc套接字连接服务器Worker：

// Frontend dealer workers
frontend, _ := zmq.NewSocket(zmq.DEALER)
defer frontend.Close()

// For workers local to the broker
backend, _ := zmq.NewSocket(zmq.DEALER)
defer backend.Close()

// Frontend should always use TCP
frontend.Bind("tcp://*:5559")

// Backend should always use inproc
backend.Bind("inproc://backend")

// Initialize Broker to transfer messages
poller := zmq.NewPoller()
poller.Add(frontend, zmq.POLLIN)
poller.Add(backend, zmq.POLLIN)

//  Switching messages between sockets
for {
    sockets, _ := poller.Poll(-1)
    for _, socket := range sockets {
        switch s := socket.Socket; s {
        case frontend:
            for {
                msg, _ := s.RecvMessage(0)
                workerID := findWorker(msg[0]) // Get server workerID from message for which it is intended
                log.Println("Forwarding Message:", msg[1], "From Client: ", msg[0], "To Worker: ")
                if more, _ := s.GetRcvmore(); more {
                    backend.SendMessage(workerID, msg, zmq.SNDMORE)
                } else {
                    backend.SendMessage(workerID, msg)
                    break
                }
            }
        case backend:
            for {
                msg, _ := s.RecvMessage(0)
                // Register new workers as they come and go
                fmt.Println("Message from backend worker: ", msg)
                clientID := findClient(msg[0]) // Get client workerID from message for which it is intended
                log.Println("Returning Message:", msg[1], "From Worker: ", msg[0], "To Client: ", clientID)
                frontend.SendMessage(clientID, msg, zmq.SNDMORE)
            }
        }
    }
}

通信流程 #

1. 客户端通过前端套接字发送包含后续消息元数据的消息：requester.SendMessage(msg)
2. 客户端发送后等待服务器确认：reply, _ := requester.RecvMessage(0)
3. Broker将消息从前端转发至对应后端Worker
4. 后端Worker处理消息后，通过后端套接字回复请求更多消息
5. Broker将回复从后端inproc转发至前端套接字
6. 客户端处理回复后，异步批量发送所需消息至服务器
7. 服务器接收并处理所有客户端消息
8. 服务器处理完成后向客户端发送ACK确认所有消息已接收
9. 服务器发送最终消息表示传输完成
10. 通信结束

队列配置 #

• Broker HWM: 10000
• Dealer HWM: 1000
• Broker Linger Limit: 0

问题相关发现 #

• 当服务器处理时长超过10分钟时，问题尤为明显
• 客户端与服务器运行在不同的Ubuntu 20LTS机器上，均使用ZMQ 4.3.2版本

环境信息 #

• libzmq版本：4.3.2
• 操作系统：Ubuntu 20LTS

排查方向建议 #

1. 套接字类型与路由逻辑检查

• Broker前端使用DEALER而非ROUTER存在逻辑缺陷：DEALER不会维护客户端的连接标识，转发回复时无法精准路由到目标客户端。客户端为DEALER时，Broker前端必须用ROUTER才能追踪每个客户端的身份，确保ACK能正确送回发起请求的客户端。
• 验证findClient和findWorker函数的映射逻辑：检查是否存在身份匹配错误，导致ACK被转发至错误客户端或因找不到目标而被丢弃。

2. 消息帧处理与SNDMORE使用规范

• 检查Broker转发时的消息帧结构：当前代码中backend.SendMessage(workerID, msg, zmq.SNDMORE)可能导致帧格式错误。ZMQ的DEALER/ROUTER消息第一帧为身份标识，后续为消息内容，需确保转发时帧数量与SNDMORE标记严格匹配，避免消息截断或冗余帧导致解析失败。
• 确认服务器Worker回复的消息结构：必须将目标客户端ID作为消息第一帧发送，否则Broker无法定位转发目标，ACK会直接丢失。

3. HWM与队列阻塞监控

• 监控队列堆积情况：当服务器处理时长超10分钟时，客户端批量发送的消息可能填满Broker或服务器队列，触发HWM阻塞发送。通过ZMQ的ZMQ_CURRENT_SND_QUEUE和ZMQ_CURRENT_RCV_QUEUE获取实时队列长度，确认是否存在队列溢出。
• 调整Linger参数：Broker Linger设为0仅影响套接字关闭时的消息处理，运行中需给客户端和服务器Worker设置合理Linger值，避免因队列阻塞导致消息无法传递。

4. 超时机制与异步逻辑优化

• 给客户端添加接收超时：当前客户端同步等待ACK（requester.RecvMessage(0)）无超时机制，一旦ACK丢失会永久阻塞。建议设置超时：requester.RecvMessage(zmq.RCVTIMEO, 30000)（30秒超时），超时后触发重试或错误处理。
• 检查服务器Worker的处理逻辑：排查是否存在线程死锁、资源耗尽（CPU、内存、文件句柄）等情况，导致无法及时发送ACK。

5. 网络与TCP链路排查

• 抓包验证ACK传输路径：使用tcpdump或wireshark抓包，确认ACK是否从服务器Worker发送到Broker、再从Broker发送到客户端。定位丢失环节：若Broker收到ACK未转发，问题在Broker逻辑；若Broker未收到ACK，问题在服务器Worker或inproc链路；若Broker发送后客户端未收到，检查TCP连接是否被防火墙断开。
• 检查网络策略：Ubuntu 20LTS的UFW是否拦截5559端口流量，或存在TCP连接超时策略，导致长时间空闲连接被断开。

6. ZMQ版本与上下文检查

• 升级libzmq版本：4.3.2存在部分DEALER/ROUTER相关已知bug，尝试升级至最新稳定版（如4.3.4）验证问题是否解决。
• 确认ZMQ上下文使用：客户端、Broker、服务器Worker需使用独立上下文，避免上下文共享导致线程安全问题。

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