Pyzmq技术问题：如何实现无阻塞发送多条消息？

从你提供的代码片段来看，问题核心出在ZeroMQ REQ/REP模式的使用规范错误以及Socket资源管理不当上，下面逐一拆解问题并给出修复方案：

核心问题点 #

• 客户端每次循环新建REQ Socket并连接：REQ/REP是严格的"请求-响应"配对模式，重复创建Socket不仅浪费资源，还会导致服务器的REP Socket在发送响应时找不到对应客户端（客户端发完消息就丢弃了Socket）。
• 使用NOBLOCK发送但未处理结果：NOBLOCK模式下如果发送缓冲区满，消息会直接发送失败，但你的代码没有任何错误检查，大概率会出现消息丢失的情况。
• 服务器Poller代码不完整：你写的P...应该是POLLIN（监听消息到达事件），而且服务器需要持续循环处理请求，同时发送响应来完成REQ/REP的配对流程。
• 客户端未处理响应：REQ模式要求发送请求后必须接收响应，否则后续操作会触发错误（虽然你每次新建Socket规避了这个问题，但这是不符合规范的用法）。

修复后的完整代码 #

客户端代码（client.py） #

import zmq
from pickle import dumps, loads

context = zmq.Context()
# 仅创建一次REQ Socket，复用连接
out_socket = context.socket(zmq.REQ)
out_socket.connect("tcp://localhost:5000")

for i in range(10):
    print(f"正在发送第{i}条消息")
    message_content = ("hello", i)
    pickled_message = dumps(message_content)
    try:
        # 去掉NOBLOCK（REQ模式默认阻塞发送，更可靠）
        out_socket.send(pickled_message)
        # 必须接收响应，完成请求-响应配对
        response = out_socket.recv()
        print(f"收到服务器响应: {loads(response)}")
    except zmq.ZMQError as e:
        print(f"发送/接收出错: {e}")

# 清理资源
out_socket.close()
context.term()

服务器代码（server.py） #

import zmq
from pickle import dumps, loads

context = zmq.Context()
in_socket = context.socket(zmq.REP)
in_socket.bind("tcp://*:5000")

poller = zmq.Poller()
# 注册监听Socket的消息到达事件
poller.register(in_socket, zmq.POLLIN)

print("服务器已启动，等待客户端消息...")
while True:
    # 轮询监听，超时时间1000ms（可按需调整）
    events = dict(poller.poll(1000))
    if in_socket in events:
        # 接收客户端消息
        pickled_message = in_socket.recv()
        message = loads(pickled_message)
        print(f"收到客户端消息: {message}")
        # 发送响应，完成配对
        response_content = ("ack", message[1])
        pickled_response = dumps(response_content)
        in_socket.send(pickled_response)
    # 可在此添加退出逻辑，比如收到特定消息后break

# 退出循环后清理资源
in_socket.close()
context.term()

关键注意事项 #

• REQ/REP的严格规则：客户端必须先发请求再收响应，服务器必须先收请求再发响应，操作顺序绝对不能颠倒。
• Socket复用：除非有特殊场景，否则不要在循环中重复创建Socket，复用已连接的Socket能大幅提升性能并避免资源泄漏。
• 异常处理：ZeroMQ的操作可能抛出ZMQError，建议添加捕获逻辑来处理发送/接收失败的情况。
• Poller使用：Poller.poll()返回事件字典，需要判断目标Socket是否触发了预期事件（比如POLLIN代表有消息可读）。

内容的提问来源于stack exchange，提问作者M.Puk