Python多线程场景下单线程异常时终止其余线程并抛出异常的实现方案

核心思路 #

• 用threading.Event作为全局停止信号，所有线程共享该信号，一旦触发所有线程主动终止运行
• 用可变容器（如列表）存储第一个抛出的异常，保证主线程可以获取到原始异常信息
• 主线程轮询停止信号状态，无需等待所有线程运行结束即可快速响应异常

改造后的自定义线程类 #

from threading import Thread

class MyThread(Thread):
    def __init__(self, stop_event, exception_container, idx, *args, **kwargs):
        super().__init__()
        self.stop_event = stop_event
        self.exception_container = exception_container
        self.idx = idx
        # 其余自定义参数初始化
        self.my_variables = ...

    def run(self):
        # 启动前已触发停止信号则直接退出
        if self.stop_event.is_set():
            return
        try:
            # 你的业务逻辑
            # 注意：在循环/耗时步骤中必须定期检查停止信号，示例如下
            # for task in your_task_list:
            #     if self.stop_event.is_set():
            #         return
            #     # 单步任务处理逻辑
            ...
        except Exception as e:
            # 仅存储第一个抛出的异常
            if not self.exception_container:
                self.exception_container.append(e)
                # 触发全局停止信号
                self.stop_event.set()

主线程实现代码 #

import threading
from your_module import MyThread

threads = []
my_list = ["a_string", "another_string", "..."]
# 全局停止信号
stop_event = threading.Event()
# 跨线程异常共享容器，用列表是因为可变对象支持跨线程修改同步
exception_container = []

# 初始化所有线程
for idx in range(len(my_list)):
    threads.append(MyThread(
        stop_event=stop_event,
        exception_container=exception_container,
        idx=idx,
        # 其余自定义参数
        ...
    ))

# 启动所有线程
for t in threads:
    t.start()

# 轮询检查异常，无需等待所有线程结束
while True:
    # 检测到异常直接抛出
    if exception_container:
        raise exception_container[0]
    # 所有线程正常运行结束则退出循环
    all_finished = all(not t.is_alive() for t in threads)
    if all_finished:
        break
    # 每100ms检查一次，降低CPU占用
    threading.Event().wait(0.1)

# 无异常时的后续业务逻辑
...

注意事项 #

Python不支持强制杀死子线程，因此必须在run方法的业务逻辑中定期检查self.stop_event.is_set()，返回True时主动结束运行。如果业务逻辑存在不可拆分的超长耗时单步操作，只能等待该步执行完成后才会响应停止信号。

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