PyZMQ异步IO：Windows服务器与Linux客户端通信响应异常排查

问题描述 #

使用PyZMQ搭建Linux客户端与Windows服务器的一对一REQ-REP连接，本地Linux环境下（客户端和服务器同机）通信正常，但跨平台部署时出现异常：客户端发送请求后偶尔无法及时收到响应，触发超时；但在服务器端终端按下回车键后，客户端会立即收到响应。响应似乎滞留在网络缓冲区，直到终端交互才被发送。

客户端精简代码 #

import typing
import zmq.asyncio
import json
import asyncio

REQUEST_TIMEOUT_S = 5.0

class Client:
    def __init__(self):
        self._context = None
        self._socket = None

    def _connect_to_remote(self, remote: str) -> typing.Tuple[zmq.asyncio.Context, zmq.asyncio.Socket]:
        # Set-up and connect Zero-MQ network socket,
        # with the `remote` being of the form `remote = f"tcp://{remoteaddr}:{remote_port}"`
        context = zmq.asyncio.Context()
        socket = context.socket(zmq.REQ)
        socket.connect(remote)
        return context, socket

    async def _send_msg(self, msg: dict):
        # Send serialized msg over zmq network socket
        serialized_msg = json.dumps(msg)
        await self._socket.send_string(serialized_msg)

    async def get_response(self, timeout_s: float = REQUEST_TIMEOUT_S):
        # Receive response from server, raises on timeout
        msg = await asyncio.wait_for(self._socket.recv(), timeout_s)
        response = json.loads(msg)

        return response

服务器精简代码 #

import zmq.asyncio
import json
import asyncio
from dataclasses import asdict

class Server:
    def __init__(self):
        self._context = None
        self._socket = None
        self._logger = None  # 假设已初始化日志实例

    def _bind_socket(self, port: int) -> tuple[zmq.asyncio.Context, zmq.asyncio.Socket]:
        # init zmq network socket and bind to port
        context = zmq.asyncio.Context()
        socket = context.socket(zmq.REP)
        bind_addr = f"tcp://*:{port}"
        socket.bind(bind_addr)
        return context, socket

    async def _rcv_msg(self):
        # Receive request from client
        # Use NOBLOCK to work around the problem that the recv must not be run into the blocking state
        # before the client sends the first msg
        message = await self._socket.recv(flags=zmq.NOBLOCK)
        deserialized_msg = json.loads(message)
        return deserialized_msg

    async def _send_msg(self, msg: dict):
        # send msg over zmq network socket
        serialized_msg = json.dumps(asdict(msg))
        await self._socket.send_string(serialized_msg)

    async def _message_loop(self):
        while True:
            try:
                message = await self._rcv_msg()
                if message is not None:
                    """
                        Do sth with the client request, here -> response
                    """
                    response = {"status": "ok", "data": "sample"}  # 示例响应
                    #  Send reply to client
                    await self._send_msg(response)
            except asyncio.CancelledError as ex:
                self._logger.exception("Cancelled messageloop")
                break
            except Exception as ex:
                self._logger.exception("Unhandled exception in messageloop")
            finally:
                await asyncio.sleep(0.1)

问题根源与解决方案 #

1. 错误使用zmq.NOBLOCK标志 #

zmq.asyncio的recv()方法本身就是异步非阻塞的，不需要额外添加zmq.NOBLOCK标志。添加该标志后，当没有消息时会立即抛出ZMQError异常，导致服务器进入频繁的异常捕获-睡眠循环，占用事件循环资源，影响响应消息的调度发送。

修复：
移除_rcv_msg中的flags=zmq.NOBLOCK，改为：

async def _rcv_msg(self):
    message = await self._socket.recv()
    deserialized_msg = json.loads(message)
    return deserialized_msg

2. 不必要的asyncio.sleep(0.1)延迟 #

finally块中的await asyncio.sleep(0.1)会强制服务器在每次循环后休眠100ms，即使已经处理完请求并发送了响应，也会延迟事件循环的调度，导致响应无法及时被推送至网络缓冲区。

修复：
删除finally块中的await asyncio.sleep(0.1)，让事件循环可以立即处理下一次请求或完成发送操作。

3. 启用TCP_NODELAY关闭Nagle算法 #

跨平台通信时，Windows和Linux的TCP栈默认行为可能存在差异，Nagle算法会合并小数据包以减少网络开销，但会导致小消息的延迟发送。启用TCP_NODELAY可以强制立即发送数据包。

修复：
在客户端和服务器创建socket后添加以下配置：

# 客户端
socket.setsockopt(zmq.TCP_NODELAY, 1)

# 服务器
socket.setsockopt(zmq.TCP_NODELAY, 1)

4. 确保Windows下asyncio事件循环正确运行 #

Python 3.8及以上版本在Windows下默认使用ProactorEventLoop，但如果你的版本较低，可能需要显式指定：

import sys
import asyncio

if __name__ == "__main__":
    if sys.platform == 'win32':
        asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
    # 启动服务器或客户端

最终验证 #

修改后重新部署服务器和客户端，测试跨平台通信，响应应能立即返回，不再出现延迟或需要终端交互触发的情况。

内容的提问来源于stack exchange，提问作者Wör Du Schnaffzig