Python中ThreadPoolExecutor遇工作线程异常时如何终止所有线程及整个脚本

你的问题核心在于：executor.map会默默捕获子线程的异常，直到所有任务执行完毕才会将异常抛出；而且在子线程里调用exit(1)只会终止当前线程，主线程还会继续等待其他任务完成。要实现"任一线程异常就立即终止所有任务并退出脚本"，可以用以下两种方案：

方案一：使用submit替代map，实时监控Future状态 #

executor.submit会返回Future对象，我们可以逐个检查这些对象的状态，一旦发现某个任务抛出异常，立即关闭线程池并终止整个程序。

修改后的代码示例：

import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

def create_dump():
    tenants = get_all_tenants()
    with ThreadPoolExecutor(max_workers=8) as executor:
        # 提交所有任务并保存Future对象
        futures = [executor.submit(process_create_dump, tenant) for tenant in tenants]
        
        # 遍历已完成的任务，实时检查异常
        for future in as_completed(futures):
            try:
                # 调用result()会触发任务中捕获的异常
                future.result()
            except Exception:
                print("Unexpected exception occurred while processing, shutting down...")
                # 关闭线程池，cancel_futures=True会取消未开始的任务
                executor.shutdown(wait=False, cancel_futures=True)
                # 终止整个脚本
                sys.exit(1)
    
    # 后续的文件操作只有在所有任务正常完成时才会执行
    with open(dump_file_path, 'a') as file:
        json.dump(json_dump, file, indent=1)

def process_create_dump(tenant):
    r_components = dict()
    print("processing.....%s" % tenant)
    # 这里不需要捕获异常，让异常向上抛给主线程处理
    add_clients(r_components, tenant)

方案二：使用共享事件（threading.Event）触发终止 #

如果需要在子线程里处理异常逻辑，可以用threading.Event作为全局信号，子线程触发异常时设置事件，主线程轮询这个事件，一旦触发就关闭线程池并退出。

代码示例：

import sys
import threading
from concurrent.futures import ThreadPoolExecutor

# 定义全局终止事件
terminate_event = threading.Event()

def create_dump():
    tenants = get_all_tenants()
    with ThreadPoolExecutor(max_workers=8) as executor:
        executor.map(process_create_dump, tenants, chunksize=1)
        
        # 轮询终止事件，一旦触发就关闭线程池
        while not terminate_event.is_set():
            # 短暂休眠避免占用过多CPU
            terminate_event.wait(timeout=0.5)
        else:
            print("Terminating due to exception in worker thread...")
            executor.shutdown(wait=False, cancel_futures=True)
            sys.exit(1)
    
    with open(dump_file_path, 'a') as file:
        json.dump(json_dump, file, indent=1)

def process_create_dump(tenant):
    # 先检查是否已经触发终止事件，避免不必要的执行
    if terminate_event.is_set():
        return
    
    r_components = dict()
    print("processing.....%s" % tenant)
    try:
        add_clients(r_components, tenant)
    except Exception:
        print("Unexpected exception occurred while processing")
        # 设置终止事件
        terminate_event.set()
        # 退出当前线程
        return

关键注意点： #

• executor.shutdown(wait=False, cancel_futures=True)：wait=False让主线程不用等待已运行的任务完成，cancel_futures=True会取消所有尚未开始执行的任务。
• 对于已经在运行的任务，Python无法强制终止线程，所以如果需要中断正在运行的任务，你需要在add_clients函数里定期检查terminate_event，比如：

  def add_clients(r_components, tenant):
      # 假设这里有循环操作
      for item in some_data:
          if terminate_event.is_set():
              raise RuntimeError("Termination requested")
          # 执行具体操作
  

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