如何使用Python multiprocessing.connection实现两个本地服务端通信

实现思路 #

两个监听不同端口的服务端要实现双向通信，需要每个节点同时承担两个角色：

• 作为服务端，监听本地端口，等待对端节点主动连接，接收对端发送的消息
• 作为客户端，主动向对端的监听端口发起连接，向对端发送消息

我们可以用线程单独跑监听逻辑，避免阻塞主动连接和消息发送的流程，同时增加连接重试逻辑兼容两个节点的任意启动顺序。

完整示例代码 #

通用逻辑（两个节点仅需修改端口配置即可） #

import threading
import time
from multiprocessing.connection import Listener, Client

# 认证密钥，两个节点必须完全一致
AUTH_KEY = b"secret password"

def run_listener(local_port: int, authkey: bytes):
    """启动本地监听，接收对端消息"""
    with Listener(("localhost", local_port), authkey=authkey) as listener:
        print(f"本地端口 {local_port} 已启动监听")
        with listener.accept() as conn:
            print(f"收到对端连接，来源地址：{listener.last_accepted}")
            # 持续监听对端发送的消息
            while True:
                try:
                    msg = conn.recv()
                    print(f"\n收到对端消息：{msg}")
                except (EOFError, ConnectionResetError):
                    print("对端连接已断开")
                    break

def connect_to_peer(peer_port: int, authkey: bytes, retry_interval: int = 2):
    """主动连接对端端口，支持重试"""
    while True:
        try:
            conn = Client(("localhost", peer_port), authkey=authkey)
            print(f"已成功连接到对端端口 {peer_port}")
            return conn
        except ConnectionRefusedError:
            print(f"对端端口 {peer_port} 未就绪，{retry_interval}秒后重试...")
            time.sleep(retry_interval)

if __name__ == "__main__":
    # --------------------------
    # 节点A配置：监听6000，连接对端7000
    LOCAL_PORT = 6000
    PEER_PORT = 7000
    # 节点B配置只需把上面两行改成下面的即可
    # LOCAL_PORT = 7000
    # PEER_PORT = 6000
    # --------------------------

    # 启动后台监听线程
    listener_thread = threading.Thread(
        target=run_listener,
        args=(LOCAL_PORT, AUTH_KEY),
        daemon=True
    )
    listener_thread.start()

    # 连接对端
    peer_conn = connect_to_peer(PEER_PORT, AUTH_KEY)

    # 主线程处理输入并发送消息
    while True:
        send_msg = input("输入要发送的内容：")
        try:
            peer_conn.send(send_msg)
        except (EOFError, BrokenPipeError):
            print("发送失败，对端连接已断开")
            break

使用方法 #

1. 把代码复制两份，第一份保留节点A的端口配置，第二份改成节点B的端口配置
2. 任意顺序启动两个程序，等待连接建立完成
3. 在任意一个程序的控制台输入内容，回车后即可发送到另一个节点，对方会自动打印收到的消息

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