如何让Python Socket实现的聊天服务稳定持续接收用户消息

问题根因 #

• 空数据判断逻辑后置：socket.recv()返回空值代表客户端已断开TCP连接，你当前代码先执行了消息转发逻辑，再判断空数据，直接导致转发时往已关闭的socket写数据抛出BrokenPipeError，处理线程直接终止
• 无效连接未清理：用户退出后没有将对应socket从在线连接列表、用户名列表中移除，后续每次消息转发都会遍历到无效socket
• 全局列表无锁保护：多线程同时读写users、sort两个全局列表会出现竞态条件，极端场景会出现列表数据错乱、增删失效
• 无异常捕获机制：所有socket操作没有捕获网络相关异常，任意连接的网络波动、异常断开都会直接终止服务端处理线程

修复方案 #

服务端完整修复代码 #

import socket
import threading
import time

HOST = '127.0.0.1'
PORT = 8888

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 加端口复用，避免重启服务时端口占用
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((HOST, PORT))

server.listen(15)
print(f'Server {HOST}:{PORT} start.')

users = [] 
online_sockets = [] # 替换原语义不明的sort变量
# 全局操作锁，保护共享列表
lock = threading.Lock()


def crypto(text, key):
    encrypt = ''
    for i in text:
        encrypt += chr(ord(i) + key)
    return encrypt   


def listen_decode(user, addr):
    print(f'User IP-address {addr[0]} login..')
    name1 = ""
    try:
        # 操作共享资源前加锁
        with lock:
            online_sockets.append(user) 
        user.send('Encode'.encode('utf-8'))
        user.send('Name'.encode('utf-8'))
        name1 = user.recv(1024).decode('utf-8')
        with lock:
            users.append(name1)

        while True:
            data = user.recv(1024).decode('utf-8')
            # 先判断连接是否断开
            if not data:
                print(f'User {name1} leave')
                break
            b1 = time.ctime()
            atribute = ' | '
            data_crypto = crypto(data, 4)
            print(f'{name1} sent message: {data_crypto} {atribute} {b1} ')

            # 转发消息，加锁遍历列表
            with lock:
                # 复制一份列表遍历，避免遍历过程中列表变化报错
                for sock in online_sockets.copy():
                    if sock != user:
                        try:
                            sock.sendall(f'{name1} > {data}'.encode('utf-8'))
                        except:
                            # 发送失败说明该连接已失效，清理掉
                            online_sockets.remove(sock)
    except Exception as e:
        print(f'User {name1 if name1 else addr[0]} connection error: {str(e)}')
    finally:
        # 无论正常退出还是异常退出，都清理资源
        with lock:
            if user in online_sockets:
                online_sockets.remove(user)
            if name1 in users:
                users.remove(name1)
        user.close()
        print(f'User {name1 if name1 else addr[0]} clean up done')


def start_server():
    while True:
       user_socket, addr = server.accept()
       potok_info = threading.Thread(target=listen_decode, args=(user_socket, addr))
       # 设为守护线程，主程序退出时自动关闭所有子线程
       potok_info.daemon = True
       potok_info.start()


if __name__ == '__main__':
    try:
        start_server()
    except KeyboardInterrupt:
        print('Server shutdown...')
        server.close()

客户端必要修改（解决异常退出问题） #

在客户端代码的window.mainloop()之前添加窗口关闭回调，主动关闭socket：

def on_close():
    try:
        client.close()
    except:
        pass
    window.destroy()

window.protocol("WM_DELETE_WINDOW", on_close)
window.mainloop()

修复说明 #

1. 调整空数据判断位置，收到数据先判断是否为空，为空直接走退出逻辑
2. 添加线程锁保护全局共享列表，所有增删改查操作都在锁范围内执行，避免竞态条件
3. 所有socket操作都加异常捕获，出现任意连接错误直接清理该连接的所有资源，不影响其他线程运行
4. 转发消息时如果单个连接发送失败，直接清理该无效连接，避免后续重复报错
5. 新增finally块保证无论线程是正常退出还是异常终止，都会清理对应连接的资源，不会残留无效socket
6. 添加端口复用、守护线程、键盘中断处理等工程化逻辑，提升服务端稳定性

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