Python线程优雅终止：如何捕获_shutdown阶段的KeyboardInterrupt异常？

嘿，我太懂你碰到的这个坑了——按下Ctrl+C终止程序时，Python在清理线程的_shutdown阶段抛出了未捕获的KeyboardInterrupt，而且普通的try-except根本抓不到，对吧？这是因为这个异常发生在主线程已经执行完所有代码之后，Python解释器在收尾清理后台线程的时候触发的，不在你之前的异常捕获范围内。

为什么你的try-except无效？ #

你之前加的try-except大概率是包裹在主线程的业务代码里，但当主线程跑完main()函数的逻辑后，Python会自动进入线程模块的清理流程：调用threading._shutdown函数，尝试join所有还在运行的非守护线程。这个阶段抛出的KeyboardInterrupt，完全避开了你之前的异常捕获逻辑，所以才会出现“捕获无效”的情况。

解决方案1：将线程设为守护线程（简单快速） #

如果你的子线程不需要在主线程退出后继续运行，也没有必须完成的收尾工作，最简单的方法就是把线程标记为守护线程。当主线程退出时，所有守护线程会被直接终止，Python不会去等待它们join，自然也就不会触发这个异常。

修改你的MCVE代码，在创建线程时加上daemon=True参数：

import threading
import time

class MCVE:
    def __init__(self):
        self.list_1 = []
        self.init()

    def thread_1(self):
        while True:
            self.list_1.append('whatever')
            time.sleep(5)

    def thread_2(self):
        while True:
            for id, entry in enumerate(self.list_1):
                print('ID {} | entry: {}'.format(id, entry))
            time.sleep(10)

    def init(self):
        # 添加daemon=True标记，将线程设为守护线程
        thread_1 = threading.Thread(target=self.thread_1, daemon=True)
        thread_2 = threading.Thread(target=self.thread_2, daemon=True)
        print('\nStarting threads...')
        print('- - - - - - - - - -\n')
        thread_1.start()
        thread_2.start()

def main():
    example = MCVE()
    # 让主线程保持运行，直到收到中断信号
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n程序已终止")

if __name__ == "__main__":
    main()

解决方案2：优雅终止线程（推荐，适合需要资源清理的场景） #

如果你的子线程需要在退出前完成收尾工作（比如保存数据、关闭网络连接），那最好用“停止标志”的方式主动控制线程退出，然后在主线程中主动join它们，避免Python进入自动清理流程。

修改后的代码示例：

import threading
import time

class MCVE:
    def __init__(self):
        self.list_1 = []
        self.running = True  # 线程运行控制标志
        self.init()

    def thread_1(self):
        # 用self.running控制循环，收到停止信号后主动退出
        while self.running:
            self.list_1.append('whatever')
            time.sleep(5)
        print("线程1已完成收尾，退出")

    def thread_2(self):
        while self.running:
            for id, entry in enumerate(self.list_1):
                print('ID {} | entry: {}'.format(id, entry))
            time.sleep(10)
        print("线程2已完成收尾，退出")

    def init(self):
        # 将线程保存为实例属性，方便后续调用join
        self.thread_1 = threading.Thread(target=self.thread_1)
        self.thread_2 = threading.Thread(target=self.thread_2)
        print('\nStarting threads...')
        print('- - - - - - - - - -\n')
        self.thread_1.start()
        self.thread_2.start()

def main():
    example = MCVE()
    try:
        # 让主线程保持运行，等待中断信号
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n收到终止信号，正在优雅退出...")
        # 设置停止标志，通知子线程退出
        example.running = False
        # 等待子线程完成收尾工作
        example.thread_1.join()
        example.thread_2.join()
        print("所有线程已退出，程序结束")

if __name__ == "__main__":
    main()

原理说明 #

第二种方法的核心逻辑是：

• 用一个可共享的布尔标志（self.running）让子线程主动退出循环，而非被强制终止
• 在主线程捕获KeyboardInterrupt后，主动通知子线程退出，然后调用join()等待它们完成收尾
• 这样Python的线程模块就不会在_shutdown阶段去处理这些线程，自然也就不会抛出那个未捕获的异常

内容的提问来源于stack exchange，提问作者G. Ramistella