如何修改Python Socket代码实现多客户端间消息转发？

核心问题分析 #

1. 客户端语法错误：send_mess函数中使用了未定义的变量dlug_wiad，导致运行报错。
2. 消息广播逻辑错误：当前服务器会把消息发给所有客户端（包括发送方），导致发送方收到自己的消息。
3. 线程安全问题：多线程操作conns列表未加锁，同时用遍历方式直接发送消息，容易出现连接失效引发的WinError 10057错误。
4. 客户端接收逻辑阻塞：发送消息后立刻调用recv，导致只能接收自己的消息，无法实时接收其他客户端的消息。

修正后的客户端代码 #

import socket
import threading

HEADER = 64
PORT = 5050
FORMAT = "UTF-8"
DISC_MESS = "END!"
SERVER = socket.gethostbyname(socket.gethostname())
ADDR = (SERVER, PORT)

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(ADDR)

def send_mess(mess):
    message = mess.encode(FORMAT)
    mess_len = len(message)
    # 修正未定义变量问题：dlug_wiad 改为 mess_len
    send_len = str(mess_len).encode(FORMAT)
    send_len += b' ' * (HEADER - len(send_len))
    client.send(send_len)
    client.send(message)

def receive_mess():
    """单独线程处理消息接收，避免阻塞输入"""
    while True:
        try:
            mess_len = client.recv(HEADER).decode(FORMAT)
            if mess_len:
                mess_len = int(mess_len)
                mess = client.recv(mess_len).decode(FORMAT)
                print(f"\n收到消息: {mess}")
                print("发送消息: ", end="")
        except:
            print("与服务器断开连接")
            client.close()
            break

# 启动接收线程
threading.Thread(target=receive_mess, daemon=True).start()

message = ""
while message != DISC_MESS:
    message = input("发送消息: ")
    send_mess(message)

修正后的服务器代码 #

import socket
import threading
import queue

HEADER = 64
PORT = 5050
SERVER = socket.gethostbyname(socket.gethostname())
ADDR = (SERVER, PORT)
FORMAT = "UTF-8"
DISC_MESS = "END!"

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(ADDR)

# 线程锁：保护连接列表的线程安全操作
conn_lock = threading.Lock()
# 存储所有活跃连接
conns = []
# 消息队列：用于线程间传递待广播的消息
msg_queue = queue.Queue()

def broadcast_thread():
    """专门的广播线程，从队列取消息并分发给除发送方外的所有客户端"""
    while True:
        mess, sender_conn = msg_queue.get()
        # 加锁遍历连接，避免并发修改问题
        with conn_lock:
            # 遍历所有连接，排除发送方
            for conn in conns.copy():
                if conn != sender_conn:
                    try:
                        # 按协议发送消息
                        message = mess.encode(FORMAT)
                        mess_len = len(message)
                        send_len = str(mess_len).encode(FORMAT)
                        send_len += b' ' * (HEADER - len(send_len))
                        conn.send(send_len)
                        conn.send(message)
                    except:
                        # 发送失败，说明客户端已断开，移除连接并关闭
                        conn.close()
                        conns.remove(conn)
        msg_queue.task_done()

def handle_client(conn, addr):
    print(f"[NEW CONNECTION] {addr} connected")
    try:
        while True:
            mess_len = conn.recv(HEADER).decode(FORMAT)
            if mess_len:
                mess_len = int(mess_len)
                mess = conn.recv(mess_len).decode(FORMAT)
                print(f"[{addr}] {mess}")
                
                if mess == DISC_MESS:
                    break
                
                # 将消息和发送方连接放入队列，交给广播线程处理
                msg_queue.put((f"[{addr}] {mess}", conn))
    finally:
        # 客户端断开，清理连接
        with conn_lock:
            if conn in conns:
                conns.remove(conn)
        conn.close()
        print(f"[DISCONNECTED] {addr} disconnected")
        print(f"[ACTIVE CONNECTIONS] {threading.active_count() - 2}")  # 减去主线程和广播线程

def start():
    server.listen()
    # 启动广播线程
    threading.Thread(target=broadcast_thread, daemon=True).start()
    print(f"[LISTENNING] Server is listening on {ADDR}")
    while True:
        conn, addr = server.accept()
        with conn_lock:
            conns.append(conn)
        thread = threading.Thread(target=handle_client, args=(conn, addr))
        thread.start()
        print(f"[ACTIVE CONNECTIONS] {threading.active_count() - 2}")  # 减去主线程和广播线程

print("Server is starting... ")
start()

关键修改说明 #

1. 客户端接收逻辑重构：

   • 新增receive_mess线程专门处理消息接收，避免输入操作被阻塞，实现实时接收其他客户端消息。
   • 修正了send_mess中的未定义变量错误。

2. 服务器线程安全与广播逻辑优化：

   • 加入conn_lock线程锁，确保多线程对conns列表的操作（添加、删除、遍历）安全，避免并发冲突。
   • 使用msg_queue消息队列，让每个客户端线程把消息放入队列，由专门的广播线程处理分发，避免直接在客户端线程中遍历发送引发的异常。
   • 广播时排除发送方连接，解决发送方收到自己消息的问题。
   • 捕获发送异常，自动移除已断开的无效连接，避免WinError 10057错误。

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