如何使用Python Socket实现P2P文本消息传输 解决同步运行报错问题

现有代码问题根因 #

原代码所有逻辑在单线程串行执行，connect()和accept()均为阻塞调用，必须等待另一个对等端完成对应操作才会返回，固定休眠的方案容错率极低，只要某一端启动慢于预设时间就会连接失败，也无法实现优雅退出。

替代实现方案 #

采用多线程拆分服务端监听、客户端连接、消息接收逻辑，客户端连接新增失败自动重试，无需固定休眠，通用对等端代码如下，启动时传入本地接收端口、对端端口即可，无需维护两份代码：

import socket
import time
import threading

def run_server(host, port_r, conn_dict):
    """服务端监听逻辑，放到独立线程运行"""
    serv_skt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 开启端口复用，避免退出后短时间内无法重启
    serv_skt.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    serv_skt.bind((host, port_r))
    serv_skt.listen(1)
    print(f"本地服务端启动，监听端口：{port_r}")
    try:
        connection, address = serv_skt.accept()
        conn_dict['server_conn'] = connection
        conn_dict['server_skt'] = serv_skt
        welcome_message = f"Welcome to {host} at {port_r}".encode('utf-8')
        connection.send(welcome_message)
        print(f"对端连接成功，地址：{address}")
    except:
        pass

def recv_msg(conn_dict):
    """收消息逻辑，放到独立线程运行，不会被输入操作阻塞"""
    while True:
        if 'server_conn' not in conn_dict:
            time.sleep(0.1)
            continue
        try:
            data = conn_dict['server_conn'].recv(1024).decode('utf-8')
            if not data:
                print("\n对端已断开连接")
                break
            print(f"\n收到消息：{data}")
            print("输入消息发送：", end="", flush=True)
        except:
            break

if __name__ == "__main__":
    HOST = '127.0.0.1'
    # 启动时输入参数：peer1输入8080 9090，peer2输入9090 8080即可
    PORT_R = int(input("输入本地接收端口："))
    PORT_S = int(input("输入对端监听端口："))
    conn_dict = {} # 用于跨线程传递连接对象

    # 启动服务端监听线程
    server_thread = threading.Thread(target=run_server, args=(HOST, PORT_R, conn_dict), daemon=True)
    server_thread.start()

    # 客户端自动重试连接，直到对端启动
    client_skt = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    print("尝试连接对端...")
    while True:
        try:
            client_skt.connect((HOST, PORT_S))
            conn_dict['client_skt'] = client_skt
            print("连接对端成功")
            break
        except ConnectionRefusedError:
            time.sleep(1)

    # 启动收消息线程
    recv_thread = threading.Thread(target=recv_msg, args=(conn_dict,), daemon=True)
    recv_thread.start()

    # 主线程处理发消息逻辑，支持优雅退出
    try:
        while True:
            msg = input("输入消息发送：")
            if msg.lower() == 'exit':
                break
            client_skt.sendall(msg.encode('utf-8'))
    except KeyboardInterrupt:
        print("\n主动退出程序")
    finally:
        # 退出前主动释放所有socket资源，避免端口占用
        if 'client_skt' in conn_dict:
            conn_dict['client_skt'].close()
        if 'server_conn' in conn_dict:
            conn_dict['server_conn'].close()
        if 'server_skt' in conn_dict:
            conn_dict['server_skt'].close()

上述方案支持：

• 无需固定休眠等待，自动重试连接直到两端都启动
• 双向自由发消息，不需要等对方回复再发送
• 按Ctrl+C或者输入exit即可优雅退出，不会出现端口占用问题

其他Socket实现P2P通信的方式 #

• UDP打洞：公网P2P场景最常用的方案，通过公网中继服务器交换两个对等端的公网IP+端口信息，双方互相发送UDP包穿透NAT设备，后续通信流量不需要经过中继，性能高
• TCP打洞：比UDP打洞复杂度高，依赖NAT设备的端口映射一致性，适合需要TCP可靠传输能力的场景，上层不需要处理丢包重传逻辑
• 中继转发P2P：打洞失败场景的兜底方案，两个对等端都连接到同一个公网中继服务器，所有消息通过服务器转发，兼容性100%，缺点是流量需要走服务器，有额外带宽成本
• 封装成熟P2P库：直接调用libp2p、pyp2p等封装好的P2P库，不需要自己处理NAT穿透、节点发现、可靠传输等复杂逻辑，只需要调用上层接口即可实现P2P通信

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