# 批量报表并行处理后如何收集失败项：用 concurrent.futures 控制并发和验收结果

## 文档明确的机制

Python 标准库 `concurrent.futures` 提供异步执行可调用对象的高层接口。线程可用 `ThreadPoolExecutor`，独立进程可用 `ProcessPoolExecutor`；它们都实现相同的 `Executor` 接口。文档也说明，该模块不适用于 WASI/WebAssembly 环境。

批量处理一组客户文件、接口下载或报表重算时，关键不是只让任务并发跑起来，而是把每个 `Future` 与原始输入关联起来。`Executor.submit(fn, *args, **kwargs)` 会调度 `fn(*args, **kwargs)`，并返回代表本次执行的 `Future`。官方示例使用字典 `future_to_url` 保存 `Future` 与 URL 的对应关系，再在任务完成时取回输入并记录成功或异常。

收集结果通常使用 `as_completed(fs, timeout=None)`：它返回一个迭代器，按照 Future 完成或取消的顺序产出。每个 Future 完成后调用 `result(timeout=None)`：成功时返回任务返回值；任务内抛出异常时，`result()` 会抛出同一个异常；超时会抛出 `TimeoutError`；若 Future 在完成前被取消，则抛出 `CancelledError`。也可以使用 `exception(timeout=None)` 查看异常；任务正常完成时它返回 `None`。

`wait(fs, timeout=None, return_when=ALL_COMPLETED)` 则返回命名二元组：`done` 集合包含等待结束前已完成或已取消的 Future，`not_done` 集合包含仍在排队或运行的 Future。`return_when` 可选 `FIRST_COMPLETED`、`FIRST_EXCEPTION`、`ALL_COMPLETED`；其中没有异常时，`FIRST_EXCEPTION` 等价于 `ALL_COMPLETED`。

`map(fn, *iterables, timeout=None, chunksize=1, buffersize=None)` 的异常时机不同：多个调用可以并发执行，但如果某个调用抛出异常，会在从返回迭代器取回该值时抛出。文档还说明，可迭代对象默认会立即收集而不是惰性收集，除非通过 3.14 新增的 `buffersize` 限制已提交但尚未产出结果的任务数量。对 `ProcessPoolExecutor`，`chunksize` 会把长迭代切分为多个块；对线程池则没有效果。

执行器应优先通过 `with` 使用。上下文管理器退出时会像 `shutdown(wait=True)` 一样等待已排队任务完成并释放资源。显式 `shutdown(wait=True, cancel_futures=False)` 后再调用 `submit()` 或 `map()` 会抛出 `RuntimeError`；`cancel_futures=True` 只取消尚未开始的任务，已完成或正在运行的任务不会因此被取消。

## 团队自行约定的接手建议

以下是团队接手约定，不是 Python 文档额外承诺的产品能力。

第一，输入与结果分离保存。建议为每个任务分配明确的业务输入，例如文件路径、客户标识、报表日期，并在提交时建立 `future → task` 映射。循环 `as_completed` 时，成功任务写入成功清单和结果位置；捕获到异常时，把原任务、异常类型、异常信息和完成阶段写入失败清单。这样失败批次可直接由原始任务重新组成，而不是从日志中反推。

第二，不要把“进程结束”当成验收。每个成功结果至少应包含可独立核验的产物信息，例如任务返回的报表路径、记录数或校验值；文档层面可确认的做法是检查任务返回值与异常，具体业务校验字段应由团队按报表格式自行定义。若没有取得证据，应标为未验收，而不是把未完成任务归入成功。

第三，按任务性质选择执行器。文档明确线程池使用线程，进程池通过独立进程避开全局解释器锁，但进程池只能执行和返回可 pickle 的对象，且 `__main__` 必须能被子进程导入；因此文档提示它不能在交互式解释器中按预期工作，REPL 中定义的函数或 `lambda` 也不应指望可用。团队可把下载、文件读写等 I/O 任务优先放入线程池，把需要独立进程的 CPU 型重算放入进程池，但这只是选择建议；并行并不保证必然提速。

第四，并发数必须显式受控。`ThreadPoolExecutor(max_workers=None)` 在 3.13 的默认值是 `min(32, (os.process_cpu_count() or 1) + 4)`；`ProcessPoolExecutor(max_workers=None)` 在 3.13 默认使用 `os.process_cpu_count()`。文档还规定进程池 `max_workers <= 0` 会抛出 `ValueError`，Windows 上不得超过 61。为了避免压满客户接口或文件系统，团队应根据任务环境设置明确上限，而不是依赖默认值推断外部系统承受能力。

第五，避免任务内部等待同一执行器的 Future。文档明确线程池中可调用对象等待另一个 Future 可能死锁；`max_workers=1` 时，任务内提交新任务并等待结果也无法完成。对进程池，文档还明确从已提交的可调用对象中调用 `Executor` 或 `Future` 方法会导致死锁。因此失败重试应由外层收集后统一生成下一批，不应在任务内部嵌套占用同一个池等待。

## 边界与总结

`ProcessPoolExecutor` 的边界必须前置审查：函数、参数和返回值需满足 `multiprocessing.Process` 对可 pickle 性的要求；入口应放在可导入模块中，并使用 `if __name__ == "__main__":` 保护启动代码。初始化函数失败时，文档说明当前排队任务及后续提交会出现 `BrokenProcessPool`；线程池初始化失败对应 `BrokenThreadPool`。

`Future.result()` 适合在逐任务处理时立即让异常浮出；`Future.exception()` 适合先收集异常对象；`as_completed()` 适合按完成顺序验收；`wait()` 适合在指定时间点划分 `done` 与 `not_done`；`map()` 代码较短，但异常要在取回对应结果时才抛出，且失败与原始下标的对应需要团队谨慎处理。

综上，稳妥的批量报表流程应是：限定并发、提交独立任务、保留 Future 与输入映射、逐个收集返回值或异常、关闭执行器，再把失败的原始输入组成可复核的下一批。文档提供的是并发执行与异常呈现机制；失败清单格式、重试批次和业务验收口径，需要团队自行约定。

## 相关工具

[AI 工作流 ROI 诊断器](/demos/ai-workflow-roi-diagnostic/index.html) · 查看工具的输入要求、输出说明与适用范围。


来源：[批量报表并行处理后如何收集失败项：用 concurrent.futures 控制并发和验收结果](https://docs.python.org/3/library/concurrent.futures.html)