调用exit()时如何安全终止长运行线程？模块线程引发进程挂起的排查与解决

你的代码出现冻结的核心原因是重复触发退出逻辑以及__del__方法在解释器退出阶段的不可靠执行：

1. 你同时通过signal.signal注册了信号处理函数、atexit.register注册了退出回调，还在__del__方法中调用了at_exit，这会导致at_exit被多次调用（比如调用exit()时，先触发atexit回调，随后全局对象销毁时又触发__del__再次调用at_exit）。
2. 在Python解释器退出阶段，__del__的执行时机和环境不稳定，此时调用线程join()可能引发未知的阻塞或死锁。
3. 你的退出标志__update_thread_alive没有加锁保护，存在多线程（信号处理线程和主线程）同时修改的竞态风险。

下面是修改后的代码，既能保证线程安全退出，又避免了重复触发和解释器冻结的问题：

from threading import Lock, Thread
import signal
import time
import atexit

class Test:
    def __init__(self):
        self.__update_thread = Thread(target=self.__update)
        self.__signals_mutex = Lock()
        self.__update_thread_alive = True
        self.__exiting = False  # 标记是否已触发退出逻辑
        
        # 注册信号处理函数
        signal.signal(signal.SIGINT, self._handle_signal)
        signal.signal(signal.SIGTERM, self._handle_signal)
        
        self.__update_thread.start()

    def _handle_signal(self, signum, frame):
        # 信号处理函数只负责触发退出，不执行阻塞操作
        self.at_exit()

    def at_exit(self):
        with self.__signals_mutex:
            if self.__exiting:
                return  # 避免重复执行退出逻辑
            self.__exiting = True
        
        print('at_exit called')
        self.__update_thread_alive = False
        self.__update_thread.join()  # 等待线程正常退出

    def __update(self):
        while self.__update_thread_alive:
            # do_stuff
            time.sleep(0.01)

__test = Test()
atexit.register(__test.at_exit)

• 移除__del__方法：atexit已经负责在解释器退出时调用清理逻辑，__del__的执行时机不可控，会导致重复触发退出操作，甚至在解释器资源已部分清理时引发问题。
• 新增__exiting标志 + 锁保护：确保at_exit只会被执行一次，避免多线程（信号处理、atexit回调）同时触发时的竞态问题。
• 拆分信号处理逻辑：信号处理函数只触发退出，不执行join()这类阻塞操作（虽然当前场景下不会死锁，但这是更安全的实践，避免信号处理过程中阻塞主线程）。
• 锁保护退出状态：使用__signals_mutex确保__exiting和__update_thread_alive的修改是线程安全的，避免多线程同时修改导致的状态不一致。

这样修改后，无论是调用exit()、按下Ctrl+C（触发SIGINT）还是收到SIGTERM信号，都能安全地通知线程退出，并等待线程完成清理，不会出现解释器冻结的问题。

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