如何在ZMQ中创建多线程ROUTER服务器以消除发送瓶颈？

我有多台DEALER客户端与ROUTER服务器进行异步通信。当前服务器通过循环检查入站请求，将请求分派给线程池worker处理，计算完成后由服务器向对应客户端返回结果，架构如下：

[client 1] <-->          <--> [worker 1]
  [client 2] <--> [server] <--> [worker 2]
  [client 3] <-->          <--> [worker 3]
  ...                           ...

但批量向客户端发送结果的速度较慢（约40ms），会拖慢服务器循环，导致无法同时处理其他客户端的结果。为消除发送瓶颈，我希望让每个worker自行将结果返回给客户端——可以直接发送，也可以通过服务器套接字，但至少不能经过Python逻辑处理。

现有代码如下：

def client():
  socket = zmq.Context.instance().socket(zmq.DEALER)
  socket.connect('tcp://localhost:4444')
  while True:
    socket.send_multipart([b'', b'request'])
    while True:
      try:
        _, response = socket.recv_multipart(zmq.NOBLOCK)
        break
      except zmq.Again:
        continue

def server():
  pool = concurrent.futures.ThreadPoolExecutor(8)
  promises = collections.deque()
  socket = zmq.Context.instance().socket(zmq.ROUTER)
  socket.bind('tcp://*:4444')
  while True:
    try:
      addr, empty, payload = socket.recv_multipart(zmq.NOBLOCK)
      promises.append(pool.submit(worker, addr, payload))
    except zmq.Again:
      time.sleep(0.0001)
    # TODO: 服务器循环里发送结果太慢
    # if promises and promises[0].done():
    #   addr, result = promises.popleft().result()
    #   socket.send_multipart([addr, b'', result])

def worker(addr, payload):
  # 执行计算任务...
  # return addr, result
  # TODO: 不想把结果返回给服务器循环，希望worker自己通过套接字发送结果

请问worker需要使用哪种套接字类型，以及如何配置它以向DEALER客户端发送消息或转发至ROUTER服务器？

方案一：Worker直接向DEALER客户端发送结果 #

• 套接字类型：使用zmq.DEALER套接字
• 配置方式：
  1. 每个worker复用全局ZMQ上下文，创建独立的DEALER套接字，连接到ROUTER服务器的地址（tcp://localhost:4444）
  2. 发送结果时，直接调用send_multipart([addr, b'', result])——这里的addr是服务器传入的客户端唯一标识，ROUTER会自动将消息路由到对应DEALER客户端
  3. 短任务场景下，发送完成后关闭套接字；长驻worker可保持套接字连接复用，减少开销

修改后的worker代码示例：

def worker(addr, payload):
    # 执行计算任务
    result = b'processed_result'
    
    # 创建DEALER套接字并发送结果
    socket = zmq.Context.instance().socket(zmq.DEALER)
    socket.connect('tcp://localhost:4444')
    socket.send_multipart([addr, b'', result])
    socket.close()

方案二：用ZMQ代理绕开Python逻辑转发（性能最优） #

如果希望完全避免Python层的转发开销，可以使用ZMQ内置的代理模式，让底层自动处理请求和结果的路由：

• 服务器端：启动zmq.ROUTER（对接客户端）和zmq.DEALER（对接worker）套接字，调用zmq.proxy实现底层自动转发
• Worker端：使用zmq.REP套接字连接到服务器的DEALER端口，处理完任务直接返回结果，代理会自动将结果转发给对应客户端

简化代码示例：

def server_proxy():
    context = zmq.Context.instance()
    # 前端对接DEALER客户端
    frontend = context.socket(zmq.ROUTER)
    frontend.bind('tcp://*:4444')
    
    # 后端对接worker
    backend = context.socket(zmq.DEALER)
    backend.bind('tcp://*:5555')
    
    # ZMQ底层自动转发，无需Python循环处理
    zmq.proxy(frontend, backend)

def worker():
    context = zmq.Context.instance()
    socket = context.socket(zmq.REP)
    socket.connect('tcp://localhost:5555')
    
    while True:
        addr, empty, payload = socket.recv_multipart()
        # 执行计算任务
        result = b'processed_result'
        # 直接返回，代理自动转发给客户端
        socket.send_multipart([addr, b'', result])

关键注意事项 #

• 复用全局ZMQ上下文：zmq.Context.instance()支持多线程安全访问，避免每个worker创建新上下文带来的资源浪费
• 套接字复用：长驻worker保持套接字连接，比每次创建关闭更高效
• 客户端兼容性：两种方案都无需修改现有DEALER客户端代码，可正常接收结果

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