Python+ZMQ实现带多worker的非阻塞REST API并行任务分发咨询

• 全局共享单个同步ZMQ REQ套接字：REQ套接字本身强制遵守发送→接收→发送的串行访问规则，同一时间只能处理一个请求，后续请求会被强制排队
• 同步ZMQ调用阻塞asyncio事件循环：原有代码中socket.recv()是同步阻塞方法，调用时会挂起整个aiohttp的事件循环，导致所有后续请求都无法被调度处理

1. API转发层修改 #

替换为asyncio兼容的异步ZMQ实现，每个请求使用独立的REQ套接字避免串行限制，同时统一用标准JSON序列化避免格式解析异常，修改后代码如下：

import json
import zmq
import zmq.asyncio
from aiohttp import web

# 初始化异步ZMQ上下文
context = zmq.asyncio.Context()
url_client = "tcp://localhost:5559"

async def process_orders(req):
    # 每个请求创建独立REQ套接字，无串行访问限制
    socket = context.socket(zmq.REQ)
    socket.connect(url_client)
    try:
        body = await req.json()
        # 直接序列化标准JSON，避免后续字符串替换风险
        await socket.send(json.dumps(body).encode())
        response = await socket.recv()
        return web.Response(text=response.decode())
    finally:
        # 请求处理完成后关闭套接字
        socket.close()

app = web.Application()
app.router.add_post('/process_orders', process_orders)

web.run_app(app, port=6666)

2. Worker端适配修改 #

原有Worker端的ROUTER+DEALER队列设备本身自带负载均衡能力，会自动将请求分配给空闲Worker，仅需要修改请求解析逻辑适配标准JSON即可：

# 原有worker函数中解析逻辑替换为如下代码，删除单引号替换操作
parsed_request = json.loads(msg)

修改完成后并发10个请求时，会先同时派发4个请求到Worker池处理，每当有Worker完成任务返回结果，就自动将队列中等待的新请求分配给空闲Worker，所有请求总耗时为ceil(10/4) * 单任务处理耗时，符合预期的并行处理效果。如果需要进一步优化性能减少套接字创建开销，可自行实现异步REQ连接池或者使用单个DEALER套接字处理多请求标识。

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