Python中如何通过线程处理客户端关停以更新连接数并回收对应线程

TCP网络连接中，当客户端主动断开、网络中断掉线、或是被强制终止进程时，服务器端对应socket的recv()调用会返回空字节串b''，或是抛出连接类异常，这是判断连接失效的核心信号。
检测到连接失效后，直接执行清理流程即可：关闭对应socket连接、更新全局连接计数、从活跃连接列表中移除条目。当线程的run()方法执行完毕后，线程会自动终止，不需要主动强制杀掉线程（Python的线程没有安全的强制终止方法，自然退出是最稳定的实现方式）。

• 模块导入错误：Python标准库的SSL模块是小写的ssl，大写导入会报模块找不到的错误
• 变量匹配错误：创建的监听socket变量是s，但后续accept连接时错写为tcpServer.accept()，会触发变量未定义错误
• 语法版本混乱：混用Python2的print语句、Python3的字符串格式化语法，无法直接运行
• 线程逻辑缺失：自定义的TestThread类没有正确继承threading.Thread，且run方法中没有处理连接断开的场景，会一直死循环占用资源
• 并发安全问题：全局存储线程、地址的列表被多线程同时读写，没有加锁保护，会出现数据错乱、列表修改崩溃的问题
• 资源泄漏：文件读写操作每次循环都重新打开，没有用上下文管理器包裹，socket连接也没有在断开后主动关闭

主服务文件 #

import socket
import ssl
import threading
import testThread

# 全局锁，保护多线程共享的连接数据
conn_lock = threading.Lock()
# 存储所有活跃的客户端处理线程
active_threads = []
# 存储所有活跃的客户端地址
active_addresses = []

# 初始化SSL上下文
context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.verify_mode = ssl.CERT_REQUIRED
context.load_cert_chain(certfile="server.crt", keyfile="server.key")
context.load_verify_locations(cafile="ca.crt")

# 初始化TCP监听socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("0.0.0.0", 8443))
s.listen(4)

print("Waiting for connections from TCP clients...")
while True:
    conn, (ip, port) = s.accept()
    print(f"Client connected: {ip}:{port}")
    conn = context.wrap_socket(conn, server_side=True)
    
    # 初始化客户端处理线程，传入锁、共享列表的引用
    new_thread = testThread.TestThread(ip, port, conn, conn_lock, active_threads, active_addresses)
    new_thread.start()
    
    # 加锁更新全局连接状态
    with conn_lock:
        active_threads.append(new_thread)
        active_addresses.append((ip, port))
        print(f"[ACTIVE CONNECTIONS] {active_addresses} | Active client count: {len(active_threads)}")

testThread.py 线程类文件 #

import threading
from datetime import datetime

class TestThread(threading.Thread):
    def __init__(self, ip, port, conn, lock, thread_list, addr_list):
        super().__init__()
        self.ip = ip
        self.port = port
        self.conn = conn
        self.lock = lock
        self.thread_list = thread_list
        self.addr_list = addr_list
        print(f"[+] New server socket thread started for {ip}:{port}")

    def run(self):
        try:
            while True:
                data = self.conn.recv(2048)
                # 收到空数据代表客户端正常断开连接
                if not data:
                    print(f"Client disconnected gracefully: {self.ip}:{self.port}")
                    break
                
                now = datetime.now()
                log_content = f"{now} - {data.decode('utf-8', errors='ignore')}\n"
                # 自动管理文件资源，写入日志
                with open("demofile.txt", "a", encoding="utf-8") as f:
                    f.write(log_content)
        except Exception as e:
            # 捕获网络中断、客户端强杀等异常场景
            print(f"Connection lost for {self.ip}:{self.port}: {str(e)}")
        finally:
            # 无论何种原因退出，都强制执行清理逻辑
            self.conn.close()
            with self.lock:
                if self in self.thread_list:
                    self.thread_list.remove(self)
                if (self.ip, self.port) in self.addr_list:
                    self.addr_list.remove((self.ip, self.port))
                remain_count = len(self.thread_list)
                print(f"[CLIENT CLOSED] Cleanup finished for {self.ip}:{self.port} | Remaining active connections: {remain_count}")

• 异常覆盖场景：无论是客户端正常断开、网络临时中断、还是客户端进程被强制终止，只要recv调用返回空值或是抛出异常，都会被捕获并进入清理流程，不需要额外的心跳机制就能覆盖基础关停场景
• 线程自动退出：所有清理逻辑都放在run()方法的finally块中，方法执行完成后线程会自动回收，不需要手动干预
• 并发安全：所有对全局共享列表的读写操作都在互斥锁的保护下执行，避免多线程同时修改列表导致的内存错误
• 资源自动释放：socket连接、文件句柄都会在流程结束时自动关闭，不会出现长期运行后的资源泄漏问题

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