主线程出现未捕获异常时，如何优雅退出子线程？

问题背景 #

你实现的MyThread类通过_stop_request事件实现优雅退出，但主线程抛出未捕获异常时，不同Python版本行为不一致：

• Python 3.5（Debian WSL）：工作线程能正常收尾
• Python 3.7（Windows 10）：工作线程持续运行且无法用Ctrl+C终止
• Python 3.9（Ubuntu 20.04）：工作线程持续运行但可通过Ctrl+C终止

设置daemon=True会直接强制终止子线程，无法执行收尾逻辑，而全局try-except又过于繁琐。

优化方案 #

方案1：自定义全局异常钩子 #

通过sys.excepthook拦截未捕获的异常，在异常处理时主动调用工作线程的join()方法，触发其收尾逻辑。

from time import sleep
from threading import Event, Thread
import sys

class MyThread(Thread):
    def __init__(self, *args, **kwargs):
        self._stop_request = Event()
        super().__init__(*args, **kwargs)
    
    def run(self):
        while not self._stop_request.is_set():
            print("My thread is running")
            sleep(.1)
        print("My thread is about to stop")  # 收尾工作
        
    def join(self, *args, **kwargs):
        self._stop_request.set()
        super().join(*args, **kwargs)

# 保存原始异常钩子
original_excepthook = sys.excepthook

def custom_excepthook(exc_type, exc_value, exc_traceback):
    # 先执行原始异常处理逻辑
    original_excepthook(exc_type, exc_value, exc_traceback)
    # 触发工作线程收尾
    if 'my_thread' in globals() and my_thread.is_alive():
        my_thread.join()

sys.excepthook = custom_excepthook

if __name__ == "__main__":
    my_thread = MyThread()
    my_thread.start()
    sleep(2)
    raise RuntimeError("Something went wrong!")

优点：无需修改原有业务代码结构，全局统一处理未捕获异常。
注意：如果有多个工作线程，需要维护线程列表在钩子中批量处理。

方案2：使用atexit注册退出清理函数 #

atexit模块会在进程正常退出或因未捕获异常退出时执行注册的清理函数，适合统一管理线程的收尾逻辑。

from time import sleep
from threading import Event, Thread
import atexit

class MyThread(Thread):
    def __init__(self, *args, **kwargs):
        self._stop_request = Event()
        super().__init__(*args, **kwargs)
    
    def run(self):
        while not self._stop_request.is_set():
            print("My thread is running")
            sleep(.1)
        print("My thread is about to stop")  # 收尾工作
        
    def join(self, *args, **kwargs):
        self._stop_request.set()
        super().join(*args, **kwargs)

def cleanup_threads():
    if 'my_thread' in globals() and my_thread.is_alive():
        my_thread.join()

atexit.register(cleanup_threads)

if __name__ == "__main__":
    my_thread = MyThread()
    my_thread.start()
    sleep(2)
    raise RuntimeError("Something went wrong!")

优点：逻辑简洁，无需处理异常类型，只要进程退出就会触发清理。
注意：如果进程被强制终止（如kill -9），atexit的函数不会执行。

方案3：封装线程管理类 #

如果有多个工作线程，可以封装一个线程管理器，统一处理线程的启动和收尾，避免重复代码。

from time import sleep
from threading import Event, Thread
import sys

class ThreadManager:
    def __init__(self):
        self._threads = []
    
    def add_thread(self, thread):
        self._threads.append(thread)
    
    def start_all(self):
        for thread in self._threads:
            thread.start()
    
    def stop_all(self):
        for thread in self._threads:
            if thread.is_alive():
                thread.join()

class MyThread(Thread):
    def __init__(self, *args, **kwargs):
        self._stop_request = Event()
        super().__init__(*args, **kwargs)
    
    def run(self):
        while not self._stop_request.is_set():
            print("My thread is running")
            sleep(.1)
        print("My thread is about to stop")  # 收尾工作
        
    def join(self, *args, **kwargs):
        self._stop_request.set()
        super().join(*args, **kwargs)

# 全局异常钩子处理
original_excepthook = sys.excepthook

def custom_excepthook(exc_type, exc_value, exc_traceback):
    original_excepthook(exc_type, exc_value, exc_traceback)
    thread_manager.stop_all()

sys.excepthook = custom_excepthook

if __name__ == "__main__":
    thread_manager = ThreadManager()
    my_thread = MyThread()
    thread_manager.add_thread(my_thread)
    thread_manager.start_all()
    sleep(2)
    raise RuntimeError("Something went wrong!")

优点：适合多线程场景，统一管理线程生命周期，扩展性强。

版本差异说明 #

Python 3.5及更早版本中，主线程抛出未捕获异常时，会等待非守护线程退出后再终止进程；但从Python 3.6开始，主线程异常退出时会直接终止所有非守护线程（Windows下存在特殊行为），这就是你看到不同版本表现不同的原因。

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