ZMQ：消息在 Dealer Router Dealer 模式实现中丢失

Question

我有一个工作设置，其中多个客户端将消息发送到多个服务器。每条消息只针对一台服务器。客户端知道所有可能的服务器的 ID，并且仅在此类服务器实际连接时才发送消息。启动时的每个服务器都连接到 socked。有多个服务器工作者绑定到 inproc 路由器套接字。通信总是从客户端发起的。消息异步发送到每个服务器。

这是使用 DEALER->ROUTER->DEALER 模式实现的。我的问题是，当客户端和服务器工作人员的数量增加时，服务器发送给客户端的“ack”（下面的第 7 步）永远不会传递给客户端。因此，客户端一直在等待确认，而服务器正在等待来自客户端的更多消息。除非重新启动，否则这两个系统都挂起并且永远不会脱离这​​种状态。配置和通信流程的详细信息如下所述。

我检查了系统日志，但没有任何明显的迹象。进一步对此进行分类的任何帮助或指导都会有所帮助。

在启动时，客户端作为经销商连接到其 IP: 端口的套接字。 “请求者，_ := zmq.NewSocket(zmq.DEALER)”。 经销商连接到经纪人。代理将前端（客户端工作人员）连接到后端（服务器工作人员）。前端绑定到 TCP 套接字，而后端绑定为 inproc。

// 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. 客户端在前端套接字上发送一组消息。消息包含有关要遵循的所有消息的元数据 请求者.SendMessage(消息)

2. 发送这些消息后，客户端等待服务器的确认 回复，_ := requester.RecvMessage(0)

3. 路由器根据上面定义的逻辑将这些消息从前端传输到后端工作人员

4. 后端经销商处理这些消息并通过后端套接字响应以请求更多消息

5. Broker 然后将消息从后端 inproc 传输到前端套接字

6. 客户端处理此消息并将所需的消息发送到服务器。消息作为一个组（批）异步发送

7. 服务器接收并处理客户端发送的所有消息

8. 处理完所有消息后，服务器向客户端发回“ack”以确认所有消息都已收到

9. 一旦所有消息都由客户端发送并由服务器处理，服务器将发送一条最终消息，指示所有传输已完成。

10. 交流到此结束

    当传输的工作人员和消息数量有限时，这非常有用。该实现有多个经销商（客户）向路由器发送消息。路由器又将这些消息发送到另一组处理相应消息的经销商（服务器）。每条消息都包含用于识别的客户端和服务器工作人员 ID。

    我们为发送和接收队列配置了以下限制。

    1. 经纪人HWM：10000
    2. 经销商 HWM：1000
    3. 经纪人逗留限制：0

       更多发现：

       1. 当服务器处理（上面的第 7 步）花费超过 10 分钟的时间时，此问题会很突出。
       2. 客户端和服务器在不同的机器上运行都是 Ubuntu-20LTS 和 ZMQ 版本 4.3.2

          环境

          • libzmq 版本（如果未发布则提交散列）：4.3.2
          • 操作系统：Ubuntu 20LTS

Answer 1

最后发现是为zmq sockets配置Heartbeat。此处参考文档http://api.zeromq.org/4-2:zmq-setsockopt 配置如下参数

1. ZMQ_HANDSHAKE_IVL：设置最大握手间隔
2. ZMQ_HEARTBEAT_IVL：设置发送ZMTP心跳的间隔
3. ZMQ_HEARTBEAT_TIMEOUT：为 ZMTP 心跳设置超时

   适当配置以上参数，以确保客户端和服务器经销商之间不断检查。因此，即使一个被延迟处理，另一个也不会突然超时。