Python concurrent收集线程池异常并关闭执行器的实现方法

使用 Python concurrent.futures 批量执行任务时，最容易漏掉的不是线程创建，而是 Future 的收口：只要不调用 future.result()，工作函数里的异常就可能一直留在 Future 中；只要不明确关闭执行器，排队任务和线程生命周期也没有清晰边界。

实用做法是把 Future 映射回任务标识，用 as_completed() 按完成顺序读取，每个结果都经过 result()，失败时记录原始异常；普通场景使用 with ThreadPoolExecutor(...) 自动等待和关闭，需要提前停止时再调用 shutdown(wait=True, cancel_futures=True)。

官方地址：https://docs.python.org/3/library/concurrent.futures.html

一、先把 Future 与任务标识绑定

不要把提交顺序当成结果顺序。线程池可能先完成后提交的任务，因此用字典保存 Future 到任务名的关系。这样日志里既有异常对象，也有输入项。

from concurrent.futures import ThreadPoolExecutor

def load_record(record_id):
    # 这里模拟一个可能失败的 I/O 任务，生产代码可替换为请求或文件读取。
    if record_id == "bad":
        raise ValueError("记录格式不正确")
    return {"id": record_id, "status": "ok"}

records = ["a-101", "bad", "a-103"]
executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="record")
future_to_record = {
    executor.submit(load_record, record_id): record_id
    for record_id in records
}

此时还没有真正收集结果。future_to_record 是后续错误归因的关键，任务函数只负责返回值或抛出异常，不要在函数内部吞掉异常。

二、用 as_completed 逐个接收结果和异常

as_completed() 返回已经完成的 Future。对每一个 Future 调用 result()，成功时取得返回值，工作函数抛错时就在当前收集点重新抛出同一个异常。

from concurrent.futures import as_completed

successes = {}
failures = {}
try:
    for future in as_completed(future_to_record):
        record_id = future_to_record[future]
        try:
            # 显式取结果，才能把工作线程中的异常带回收集线程。
            successes[record_id] = future.result()
        except Exception as exc:
            # 保存任务标识和异常类型，便于重试或告警归因。
            failures[record_id] = f"{type(exc).__name__}: {exc}"
finally:
    # wait=True 保证已开始的任务完成后再释放线程池资源。
    executor.shutdown(wait=True)

print("成功:", len(successes), "失败:", failures)

这个版本适合“尽量完成整批，再汇总失败”的业务。finally 放关闭逻辑，可以覆盖收集代码自身发生异常的情况。

三、失败后取消排队任务并关闭执行器

如果第一条关键任务失败就不想继续扩大批次，可在异常分支调用 shutdown(wait=True, cancel_futures=True)。它会取消尚未启动的 Future，但不会中断已经运行的函数，所以业务函数仍应支持自己的超时或取消标记。

executor = ThreadPoolExecutor(max_workers=4)
futures = [executor.submit(load_record, item) for item in records]
try:
    for future in as_completed(futures):
        # 先取结果；关键任务失败时让异常进入统一收口逻辑。
        future.result()
except Exception as exc:
    # 只取消尚未开始的任务，已运行任务会在 wait=True 时完成。
    executor.shutdown(wait=True, cancel_futures=True)
    raise RuntimeError("批次中止") from exc
else:
    # 全部成功时同样显式关闭，避免依赖进程退出清理。
    executor.shutdown(wait=True)

若项目只需“每项独立报告”，不要在循环里遇到第一条异常就中止；若任务有副作用，则更要先区分“已运行”和“待排队”，不能把取消理解成回滚。

四、正常批次优先使用 with 自动收口

没有提前停止需求时，上下文管理器更不容易遗漏关闭动作。它离开代码块时会按等待式关闭执行器；同时，建议把异常收集写在块内，让成功、失败和资源释放各自职责清楚。

with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交和收集都在同一生命周期内，离开 with 后不再提交新任务。
    future_to_record = {
        executor.submit(load_record, item): item for item in records
    }
    for future in as_completed(future_to_record):
        item = future_to_record[future]
        try:
            print(item, future.result())
        except Exception as exc:
            print(item, "失败:", exc)

相关问题

为什么只调用 submit 不会自动打印线程池异常？

异常被保存到 Future，调用 result() 时才会在当前线程重新抛出；不读取结果就没有收集动作。

shutdown(wait=False) 会立即结束 Python 进程吗？

不会。它可以让调用点先返回，但程序仍会等待已提交的 Future 完成，不能当作强制终止。

Future.cancel() 和 cancel_futures 有什么区别？

前者针对单个 Future，后者在执行器关闭时批量取消未启动任务；两者都不能取消已经运行的工作。