多线程聊天应用中Input函数阻塞及连接终止异常问题

问题根源 #

你的问题并非由GIL导致，核心矛盾在于：

• send()线程中的input()是阻塞式调用，即使stop_thread事件被清除，线程仍会卡在等待用户输入的状态，无法及时退出。
• 连接关闭后，客户端接收线程未处理recv()抛出的连接异常，导致程序报错。
• 重复执行socket关闭操作，可能引发额外异常。

解决方案 #

针对上述问题，我们可以通过以下方式修复：

1. 使用select模块实现非阻塞输入，让send()线程能周期性检测终止信号，避免被input()永久阻塞。
2. 统一线程退出逻辑，所有线程循环优先检测stop_thread事件。
3. 给socket操作添加异常捕获，处理连接关闭后的各类错误。
4. 避免重复关闭socket，用线程锁保护关闭操作。

修改后的服务端代码 #

import threading
from threading import Event, Lock
import socket
import select
import sys

server_socket = socket.socket()
ip = socket.gethostbyname(socket.gethostname())
port = 1234
server_socket.bind((ip, port))
server_socket.listen()
print("\n服务器正在监听客户端连接!!! ")
client_socket, client_add = server_socket.accept()
print(f"已连接到 {client_add}") 

# 线程同步工具
stop_thread = Event()
socket_lock = Lock()
is_connected = True

def send():
    global is_connected
    b = lambda x: "\n".join([x]*g)
    g = 1  # 默认重复次数
    while stop_thread.is_set():
        # 使用select实现非阻塞输入，每隔0.5秒检测一次终止信号
        ready, _, _ = select.select([sys.stdin], [], [], 0.5)
        if ready:
            sms = sys.stdin.readline().strip()
            if not sms:
                continue
            try:
                inputs = sms.split()
                getdata = inputs[0]
                if len(inputs) > 1:
                    g = int(inputs[1])
                    sms = b(getdata)
                
                with socket_lock:
                    if is_connected:
                        client_socket.send(sms.encode())
            except ValueError:
                with socket_lock:
                    if is_connected:
                        client_socket.send(sms.encode())
            except (BrokenPipeError, ConnectionResetError):
                print("连接已关闭，无法发送消息")
                break
            
            if sms.lower() == "bye":
                print("\n你主动关闭了连接!!")
                close_all_connections()
                break
        # 每次循环都检查终止信号
        if not stop_thread.is_set():
            break

def recieve():
    global is_connected
    while stop_thread.is_set():
        try:
            msg = client_socket.recv(1024)
            if not msg:
                # 客户端主动关闭连接
                print("\n客户端已断开连接!! ")
                close_all_connections()
                break
            decoded_msg = msg.decode().lower().strip()
            if decoded_msg == "bye":
                print("\n客户端请求关闭连接!! ")
                close_all_connections()
                break
            else:
                print("\n>>客户端 --> ", msg.decode())
        except (ConnectionResetError, OSError):
            print("\n连接已意外关闭")
            close_all_connections()
            break

def close_all_connections():
    global is_connected
    with socket_lock:
        if is_connected:
            is_connected = False
            stop_thread.clear()
            try:
                client_socket.close()
                server_socket.close()
            except Exception as e:
                print(f"关闭连接时出错: {e}")

# 启动线程
stop_thread.set()
t1 = threading.Thread(target=send)
t2 = threading.Thread(target=recieve)
t1.start()
t2.start()

# 等待线程结束
t1.join()
t2.join()
print("服务器已退出")

关键改动说明 #

• 非阻塞输入：用select.select监听标准输入，每隔0.5秒检测一次终止信号，解决input()阻塞导致线程无法及时退出的问题。
• 连接状态管理：用is_connected标记连接状态，配合socket_lock避免重复关闭socket或在关闭后继续发送数据。
• 异常处理：捕获BrokenPipeError、ConnectionResetError等连接相关异常，避免程序崩溃。
• 统一关闭逻辑：将连接关闭逻辑封装到close_all_connections函数，确保所有线程能同步响应终止信号。

客户端优化提示 #

客户端接收线程也需要添加类似的异常处理，当检测到连接关闭时，立即退出线程：

# 客户端接收线程示例
def recieve():
    while stop_thread.is_set():
        try:
            msg = client_socket.recv(1024)
            if not msg:
                print("\n服务器已断开连接")
                close_connection()
                break
            # 处理消息逻辑
        except (ConnectionResetError, OSError):
            print("\n连接已关闭")
            close_connection()
            break

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