Python Socket开发聊天程序触发BrokenPipeError 32错误怎么修复

问题根因 #

• 在线用户socket全局列表sort未清理已断开的连接：用户退出时仅终止了当前用户的监听线程，没有把失效socket从sort中移除，后续其他用户发消息时，服务端会遍历所有sort中的socket发送数据，向已断开的socket写数据就会触发Broken pipe错误，且没有异常捕获会直接终止当前发消息的线程，导致服务端不再处理该用户后续的消息。
• 空数据判断逻辑顺序错误：原代码先处理、群发收到的消息，再判断data是否为空，用户断开时recv返回空字节，会先执行无效的群发逻辑才退出，进一步提升了报错概率。

修复方案 #

1. 调整空数据判断位置，增加退出清理逻辑 #

listen_decode函数里，recv拿到数据后立刻判断是否为空，为空时先把当前socket和用户名从全局列表移除，再关闭socket退出线程，避免后续无效操作。

2. 群发消息增加异常捕获 #

遍历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.bind((HOST, PORT))

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

users = [] 
sort = []

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..')
    sort.append(user) 
    user.send('Encode'.encode('utf-8'))
    user.send('Name'.encode('utf-8'))
    name1 = user.recv(1024).decode('utf-8')
    users.append(name1)
    
    while True:
        # 接收数据后优先判断连接是否断开
        data = user.recv(1024).decode('utf-8')
        if not data:
            print(f'User {name1} leave')
            # 退出前清理全局列表中的无效数据
            if user in sort:
                sort.remove(user)
            if name1 in users:
                users.remove(name1)
            # 关闭失效socket释放资源
            user.close()
            break

        b1 = time.ctime()
        atribute = ' | '
        data_crypto = crypto(data, 4)
        print(f'{name1} sent message: {data_crypto} ' + atribute + '' + b1 + ' ')

        # 群发消息增加异常捕获，避免单个连接失效影响全局
        to_remove = []
        for i in sort:
            if(i != server and i != user):
                try:
                    i.sendall(f'{name1} > {data}'.encode('utf-8'))
                except:
                    # 标记发送失败的失效socket
                    to_remove.append(i)
        # 批量清理失效socket
        for invalid_sock in to_remove:
            if invalid_sock in sort:
                sort.remove(invalid_sock)


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__':
    start_server()

额外优化建议 #

• 客户端退出时可以主动给服务端发一个约定好的退出标记，而不是直接断开连接，服务端收到标记后走清理逻辑，异常处理更可控。
• 可以给每个监听线程绑定用户名标识，后续排查多用户在线问题更方便。

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