使用asyncio实现ZeroMQ客户端服务端并发时无限等待如何解决

1. 套接字类型与模式不匹配 #

• 普通REQ/REP模式不支持多服务端实例绑定同一端口，如果你启动的两个服务端监听同一端口，只有其中一个实例能正常接收消息，另一个会永久阻塞。如果需要多服务端负载，必须改用ROUTER/DEALER组合模式。
• REQ/REP模式有严格的调用时序要求：服务端必须严格遵循接收请求→返回响应的顺序，客户端必须严格遵循发送请求→等待响应的顺序，任意一端时序错误都会直接导致阻塞。

2. ZeroMQ asyncio适配错误 #

• 你大概率使用了默认同步Context而非asyncio专属Context，导致await receive()本质是同步调用，直接卡住整个事件循环。必须用zmq.asyncio.Context()创建上下文，相关初始化参考示例：

import zmq
import zmq.asyncio

class Server:
    def __init__(self, port):
        # 必须使用asyncio专属上下文
        self.ctx = zmq.asyncio.Context.instance()
        self.socket = self.ctx.socket(zmq.REP)
        self.socket.bind(f"tcp://*:{port}")
    
    async def run(self):
        while True:
            msg = await self.socket.recv()
            await self.socket.send(msg)

3. 连接与调度逻辑错误 #

• 如果两个服务端绑定了不同端口，客户端仅连接了其中一个端口，未被连接的服务端自然收不到任何消息，会卡在接收调用上。
• 协程调度时如果没有用asyncio.gather()同时启动所有服务端、客户端协程，可能出现客户端协程未被实际调度的情况，服务端自然收不到请求。调度逻辑参考示例：

async def main():
    s1 = Server(5555)
    s2 = Server(5556)
    # 客户端需要按业务需求连接对应服务端端口
    c = Client(target_addr="tcp://localhost:5555")
    # 同时调度所有协程
    await asyncio.gather(s1.run(), s2.run(), c.run())

if __name__ == "__main__":
    asyncio.run(main())

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