如何在单台PC上实现ZeroMQ双向通信？是否需线程支持？

阿华AIGC实验室

2026-5-27

问题描述

我手头有ZeroMQ的服务端和客户端代码如下：

服务端代码

import time
import zmq

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

while True:
    message = socket.recv()
    print("Received request: %s" % message)
    time.sleep(1)
    socket.send(b"World")

客户端代码

import zmq

context = zmq.Context()
print("Connecting to hello world server…")
socket = context.socket(zmq.REQ)
socket.connect("tcp://localhost:5555")

# Do 10 requests, waiting each time for a response
for request in range(10):
    print("Sending request %s …" % request)
    socket.send(b"Hello")
    # Get the reply.
    message = socket.recv()
    print("Received reply %s [ %s ]" % (request, message))

现在我需要在单台PC上实现双向并发通信：这台电脑要能同时主动发送数据，还要能接收并处理来自其他电脑的数据。想请教几个问题：

这种需求该怎么实现？
是否需要借助线程？
那个异步服务端的方案能不能解决我的问题？

解答与实现方案

1. 是否需要线程？

不一定非用线程——ZeroMQ本身支持异步IO模式，可以在单线程里处理多IO操作。但如果你的发送、接收逻辑比较复杂，或者想让两者完全解耦，用线程/多进程也是非常合理的选择，两种方式各有优劣，看你需求的复杂度。

2. 双向并发通信的具体实现

首先得明确：你现在用的REQ/REP套接字是严格的请求-响应配对模式，只能单向一问一答，完全没法满足同时收发的需求。必须换成支持全双工的套接字组合，这里推荐DEALER/ROUTER组合，这是ZeroMQ里实现双向并发通信的标准方案。

方案：同时启动ROUTER（接收端）和DEALER（发送端）

每台机器都需要同时创建两种套接字：

ROUTER：绑定本地端口，用来监听其他机器的连接，接收并处理外来消息；
DEALER：连接到目标机器的ROUTER端口，用来主动发送消息。

搭配线程（或者异步IO）就能实现同时收发，下面是一个可运行的示例代码：

import zmq
import threading

def handle_incoming(router_socket):
    """处理收到的消息，运行在单独线程"""
    while True:
        # ROUTER接收的消息格式是 [客户端标识, 空帧, 实际消息]
        client_id, _, msg = router_socket.recv_multipart()
        print(f"收到来自 {client_id.hex()} 的消息: {msg.decode()}")
        # 可选：回复消息给发送方
        router_socket.send_multipart([client_id, b"", b"已收到: " + msg])

def handle_outgoing(dealer_socket):
    """处理发送消息，运行在单独线程"""
    while True:
        user_input = input("请输入要发送的消息: ")
        dealer_socket.send_string(user_input)
        # 可选：接收对方的回复
        reply = dealer_socket.recv_string()
        print(f"收到回复: {reply}")

if __name__ == "__main__":
    context = zmq.Context()

    # 初始化ROUTER，绑定本地5555端口
    router = context.socket(zmq.ROUTER)
    router.bind("tcp://*:5555")

    # 初始化DEALER，连接到目标机器的ROUTER端口（测试时可以用localhost，实际换成其他电脑IP）
    dealer = context.socket(zmq.DEALER)
    dealer.connect("tcp://localhost:5555")

    # 启动两个线程分别处理收发
    recv_thread = threading.Thread(target=handle_incoming, args=(router,), daemon=True)
    send_thread = threading.Thread(target=handle_outgoing, args=(dealer,), daemon=True)

    recv_thread.start()
    send_thread.start()

    # 主线程保持存活，等待用户输入退出
    input("按回车键退出...\n")

如果不想用线程，也可以用ZeroMQ结合asyncio的异步方案，用协程来处理并发，代码结构类似，但不需要额外开线程，更轻量。