如何可靠释放Python线程中Listener绑定端口并优化停止检查逻辑？

1. 可靠释放端口并复用的实现方案 #

端口出现“Address already in use”的核心原因有两个：一是TCP连接关闭后的TIME_WAIT状态会占用端口一段时间；二是原代码的停止逻辑存在线程安全问题，导致Listener未被正确关闭。以下是修复后的可靠实现：

修改后的代码 #

import socket
import threading
import time
from multiprocessing.connection import Listener

class MyListener(threading.Thread):
    def __init__(self, name):
        super().__init__(daemon=False)
        self.name = name
        self.address = ('localhost', 5000)
        # 用线程安全的Event代替布尔变量
        self.stop_event = threading.Event()
        self.print("Object constructed")

    def run(self):
        self.print("Thread started")
        # 设置SO_REUSEADDR选项，允许端口复用
        listener = Listener(self.address, socket_options=((socket.SOL_SOCKET, socket.SO_REUSEADDR, 1),))
        try:
            self.print("Listener created")
            # 循环监听，直到收到停止信号
            while not self.stop_event.is_set():
                # 设置超时，避免一直阻塞在accept，定期检查停止信号
                listener._socket.settimeout(1)
                try:
                    conn = listener.accept()
                except socket.timeout:
                    continue
                try:
                    if self.stop_event.is_set():
                        break
                    msg = conn.recv()
                finally:
                    conn.close()
        except Exception as e:
            self.print(f"Listener error: {e}")
        finally:
            listener.close()
            self.print("Listener closed")
            self.print("exit run")

    def close(self):
        self.print("Closing")
        self.stop_event.set()
        # 等待线程退出，确保资源完全释放
        self.join()
        self.print("All closed")

    def print(self, text):
        thread_name = threading.current_thread().name
        thread_name = "M" if thread_name == "MainThread" else thread_name.split('-')[0]
        print(f"{thread_name}:{self.name}: {text}")

# 测试代码
sleepseconds = 1
l1 = MyListener(name="A")
time.sleep(sleepseconds)
l1.start()
time.sleep(sleepseconds)
l1.close()
time.sleep(sleepseconds)

l2 = MyListener(name="B")
time.sleep(sleepseconds)
l2.start()
time.sleep(sleepseconds)
l2.close()

关键修复点 #

• 端口复用设置：创建Listener时传入socket_options=((socket.SOL_SOCKET, socket.SO_REUSEADDR, 1),)，让操作系统允许立即复用处于TIME_WAIT状态的端口。
• 线程安全的停止机制：用threading.Event替代self.stopped布尔变量，Event的is_set()和set()方法是线程安全的，避免多线程下的竞态条件。
• 打破accept阻塞：给Listener的socket设置超时，这样线程不会一直卡在accept()调用，定期检查停止信号；调用close()时通过join()等待线程完全退出，确保资源释放完成。
• 简化资源管理：移除冗余的gc.collect()和手动删除属性的操作，通过try/finally确保连接和Listener被正确关闭。

2. 更优雅的停止方式：用threading.Event替代循环检查 #

原代码中重复检查self.stopped的方式不仅冗余，还存在线程安全隐患。使用threading.Event是更优雅的方案：

• Event是专门为线程间通信设计的同步原语，天然支持线程安全的状态通知。
• 可以结合socket超时，让线程在等待连接的间隙自动检查停止信号，无需写大量重复的条件判断。
• 调用stop_event.set()可以立即触发停止逻辑，配合join()等待线程退出，确保资源完全释放。

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