使用Python的concurrent.futures实现任务失败时取消所有任务

嘿，我完全懂你想要实现的需求——只要任务池里有任何一个任务报错，就立刻叫停所有还在运行或者没启动的任务，并且把第一个触发失败的异常抛出来。我之前处理批量任务的时候也遇到过类似的场景，咱们一步步来解决这个问题。

首先得明确一个关键限制：ProcessPoolExecutor里已经在运行的任务没法直接用cancel()取消，因为子进程和主进程是完全独立的，Python的concurrent.futures没有提供原生的强制终止子进程的方法；但ThreadPoolExecutor相对灵活，未启动的任务可以直接取消，已运行的线程如果能配合检查终止信号，也能主动退出。下面分别针对两种场景给出解决方案。

针对ProcessPoolExecutor的实现方案 #

对于进程池，我们可以通过共享终止标志让运行中的任务主动退出，同时在捕获到第一个异常后，立刻关闭executor并手动终止子进程，最后抛出第一个异常。

完整代码示例：

from concurrent.futures import ProcessPoolExecutor, as_completed
from functools import partial
import os
import signal
from multiprocessing import Event

def copy_from(file_path, table_name, column_string, stop_flag=None):
    try:
        # 这里是你的实际业务逻辑，建议加入终止标志检查
        # 示例：处理文件时每隔一段逻辑就检查是否需要终止
        # for data_chunk in read_file_chunks(file_path):
        #     if stop_flag and stop_flag.is_set():
        #         raise RuntimeError("任务因全局终止信号中断")
        #     # 执行复制操作

        # 模拟某个任务失败的情况
        if "error" in file_path:
            raise ValueError(f"处理文件 {file_path} 时发生错误")
        print(f"成功完成文件 {file_path} 的复制")
    except Exception:
        # 触发全局终止信号，通知其他任务
        if stop_flag:
            stop_flag.set()
        raise

def main():
    table_name = "target_table"
    column_string = "id, name, create_time"
    file_path_list = ["file1.txt", "file2_error.txt", "file3.txt", "file4.txt"]
    cores_to_use = 2

    # 创建跨进程共享的终止标志
    stop_flag = Event()
    copy_func = partial(copy_from, table_name=table_name, column_string=column_string, stop_flag=stop_flag)
    first_exception = None

    with ProcessPoolExecutor(max_workers=cores_to_use) as executor:
        futures = {executor.submit(copy_func, path): path for path in file_path_list}
        
        try:
            for future in as_completed(futures):
                try:
                    future.result()
                    # 如果已经收到终止信号，直接退出循环
                    if stop_flag.is_set():
                        break
                except Exception as e:
                    # 只记录第一个触发的异常
                    if not first_exception:
                        first_exception = e
                    # 触发全局终止信号
                    stop_flag.set()
                    # 立刻关闭executor，不再接受新任务
                    executor.shutdown(wait=False)
                    # 手动终止所有子进程（粗暴但有效，若需要优雅退出请依赖任务内的标志检查）
                    for pid in executor._processes.keys():
                        try:
                            os.kill(pid, signal.SIGTERM)
                        except ProcessLookupError:
                            pass
                    break
        finally:
            # 确保资源被清理
            if not executor._shutdown:
                executor.shutdown(wait=False)
                for pid in executor._processes.keys():
                    try:
                        os.kill(pid, signal.SIGTERM)
                    except ProcessLookupError:
                        pass

    # 抛出第一个捕获到的异常
    if first_exception:
        raise first_exception

if __name__ == "__main__":
    main()

代码关键点说明

• 共享终止标志：用multiprocessing.Event创建跨进程可见的终止信号，任务函数可以定期检查该标志，主动退出以实现优雅终止。
• 异常捕获逻辑：只保存第一个触发的异常，避免后续异常覆盖原始错误信息。
• 强制终止子进程：调用executor.shutdown(wait=False)阻止新任务提交后，通过os.kill给子进程发送终止信号，快速停止运行中的任务。

针对ThreadPoolExecutor的优化实现 #

线程池的场景更灵活，未启动的任务可以直接用future.cancel()取消，已运行的线程通过threading.Event共享终止信号即可：

from concurrent.futures import ThreadPoolExecutor, as_completed
from functools import partial
import threading

def copy_from(file_path, table_name, column_string, stop_flag=None):
    try:
        # 实际业务逻辑中加入终止标志检查
        # for data_chunk in read_file_chunks(file_path):
        #     if stop_flag and stop_flag.is_set():
        #         raise RuntimeError("任务因全局终止信号中断")
        #     # 执行复制操作

        if "error" in file_path:
            raise ValueError(f"处理文件 {file_path} 时发生错误")
        print(f"成功完成文件 {file_path} 的复制")
    except Exception:
        if stop_flag:
            stop_flag.set()
        raise

def main():
    table_name = "target_table"
    column_string = "id, name, create_time"
    file_path_list = ["file1.txt", "file2_error.txt", "file3.txt", "file4.txt"]
    cores_to_use = 2

    stop_flag = threading.Event()
    copy_func = partial(copy_from, table_name=table_name, column_string=column_string, stop_flag=stop_flag)
    first_exception = None

    with ThreadPoolExecutor(max_workers=cores_to_use) as executor:
        futures = {executor.submit(copy_func, path): path for path in file_path_list}
        
        try:
            for future in as_completed(futures):
                try:
                    future.result()
                    if stop_flag.is_set():
                        break
                except Exception as e:
                    if not first_exception:
                        first_exception = e
                    stop_flag.set()
                    # 取消所有未启动的任务
                    for fut in futures:
                        if not fut.done():
                            fut.cancel()
                    executor.shutdown(wait=False)
                    break
        finally:
            if not executor._shutdown:
                executor.shutdown(wait=False)

    if first_exception:
        raise first_exception

if __name__ == "__main__":
    main()

核心注意事项 #

• 进程池的优雅终止：强制杀死子进程可能导致资源泄漏（如未关闭的文件句柄），建议优先让任务函数主动检查终止标志并清理资源后退出。
• 异常优先级：一定要保证第一个触发的异常被保留并抛出，避免后续异常覆盖原始错误。
• 资源清理：在finally块中确保executor被正确关闭，防止资源泄漏。

备注：内容来源于stack exchange，提问作者Purushottam Nawale