Python as_completed 收集部分结果：单项失败与等待超时怎样区分

并发检查十个文件，只要一个Future抛错，就直接从循环返回，前面已经成功的结果也一起丢了。另一种错误更隐蔽：收集超时以后把剩下项目全部标成失败，却没有区分它们还在运行、排队等待，还是根本没有开始。

as_completed() 适合按完成次序接收Future，让调用方逐项保存结果。但它只改变结果等待的组织方式，不会替你选择部分成功政策，也不会在超时时自动停止执行。

下面用 Python 3.12 固定三项任务的状态：第一项成功，第二项产生业务错误，第三项一直等待放行。先让前两项确定结束，并等第三项发出已经开始的信号，再启动收集，保证最后的超时报告不是因为机器调度慢偶然漏掉了一项。

from concurrent.futures import ThreadPoolExecutor, as_completed, wait
from threading import Event


def inspect(ident, gate, fail=False, started=None):
    if started is not None:
        started.set()
    gate.wait()
    if fail:
        raise ValueError(f"item {ident} rejected")
    return {"id": ident, "checked": True}


def main():
    fast = Event()
    rejected = Event()
    slow = Event()
    slow_started = Event()
    pool = ThreadPoolExecutor(max_workers=3)
    futures = {}
    report = {}
    try:
        first = pool.submit(inspect, 1, fast)
        second = pool.submit(inspect, 2, rejected, True)
        third = pool.submit(inspect, 3, slow, False, slow_started)
        futures = {first: 1, second: 2, third: 3}
        fast.set()
        rejected.set()
        wait([first, second])
        if not slow_started.wait(timeout=3):
            raise AssertionError("slow task did not start")

        try:
            for future in as_completed(futures, timeout=0.05):
                ident = futures[future]
                try:
                    result = future.result()
                except ValueError as error:
                    report[ident] = {"state": "failed", "error": str(error)}
                else:
                    report[ident] = {"state": "succeeded", "result": result}
        except TimeoutError:
            for future, ident in futures.items():
                if ident not in report:
                    report[ident] = {
                        "state": "pending_at_deadline",
                        "running": future.running(),
                    }
        assert report[1]["state"] == "succeeded"
        assert report[2]["state"] == "failed"
        assert report[3]["state"] == "pending_at_deadline"
        assert not third.done()
        print("collection snapshot:", report)
    finally:
        slow.set()
        pool.shutdown(wait=True, cancel_futures=True)
    assert third.result() == {"id": 3, "checked": True}
    print("slow task finished later; the deadline snapshot is unchanged")


if __name__ == "__main__":
    main()

保存为 collect_completed.py，执行 python3 collect_completed.py。输出中的成功与失败项目按编号保留，第三项在截止时是未完成状态；清理阶段放行以后，它仍然能够完成。这两条记录说的是不同观察时点，不应互相覆盖成一个含糊的“最后好像成功了”。

完成次序与输入次序，分别由不同结构表达

Future到项目编号的映射必须在提交时建立。不能按拿到结果的循环下标回填输入数组，先结束的并不一定是第一个输入。结果里恰好带了文件名，也不能代替稳定编号，因为输入可能有重复路径或相同展示名。

已经完成的多个Future，返回顺序也不应被当成严格的业务时间线。它们在开始收集以前已经都完成，集合如何交给迭代器，并不能证明谁先写入或谁先失败。需要因果诊断时，应在执行者内部记录相应阶段和标识。

同一个Future被重复放入as_completed输入，只会被产出一次。因此，重复提交的业务项与重复登记的同一个Future又是不同问题。报告总数应来自明确的任务账本，而不是简单按迭代次数推断输入数量。

例子保留了report字典，再按需要输出。若外部合同要求输入顺序，可以在收集完成后根据输入编号排序；若界面要及时展示进度，则每收到一项就更新对应行，但仍使用编号定位，不能让完成次序改变项目身份。

迭代器超时和单项执行异常不要混成一类

Future自己的异常由result抛出，收集等待超时则发生在as_completed迭代过程中。代码分别捕获，才知道是某项被拒绝，还是调用方不再继续等待剩余项。

timeout也不是每次产出一个结果以后重新给一份完整预算。它约束这一轮等待过程，不能靠不断收到快任务让无限慢任务永远等下去。若需要每项独立执行预算，应让工作函数或其下游操作分别支持，而不是在result外面反复套相同时间参数。

超时发生时，某项可能已经完成但尚未被收集。示例的前两项预先等到完成，便于固定基础状态；实际业务如果要做严谨截止快照，应继续检查未登记Future的done、cancelled与异常状态，定义好同时完成时归入哪边。不要把“还没在循环里处理”直接等同于“执行失败”。

当前程序把未收集项统一标记pending_at_deadline，是一份明确的观察报告，不是最终任务结论。产品若需要最终报告，就应由仍然拥有线程池的组件稍后继续收集，并保留状态更新时间。

任务返回值为空也不应该被当作失败。None可能是一个合法成功结果，异常与取消则是Future的独立状态。报告结构使用state字段，而不是用result是否为真来判定，能够保留零、空列表和None这些有效业务值。

同样，记录错误时应保留稳定项目编号和错误类型。直接把完整异常正文发给公开界面，可能夹带本地路径或下游敏感响应；对外报告与内部诊断可以不同，但都需要关联到同一项任务。部分成功的价值不只是留下几条数据，还在于每一项最终状态能够被解释。

取消只能作为下一步动作，不能伪装成已经发生的结果

排队但未开始的Future可能取消成功，正在执行的普通线程任务通常不能靠Future.cancel强制中断。取消请求与取消成功要区分，更不能在发出请求后立刻标记“已停止”。

工作函数若有网络调用，应传递其自身支持的超时或停止信号；本例使用Event，是为了展示调度状态，不是建议业务把所有任务都挂在一个全局开关上。每个批次的退出责任应由相应组件拥有。

finally中放行slow并shutdown等待，使脚本退出时没有遗留工作。若调用方必须在截止点立即返回，就需要把线程池与剩余Future交给更长寿命的管理者。不能删掉finally以后丢掉引用，仍宣称批次已经完整结束。

这份报告还允许部分成功，但没有自动重试。失败项能否重试取决于错误类型，未完成项更需要核实实际状态；写入结果未知时贸然重交，可能制造重复副作用。

把第三项的Event提前放行，再运行一次，可以得到没有收集超时的版本。随后再把第二项的fail改为False，应得到三项成功。保留这两份变化与原来的部分完成样本，能分别验证完成收集、单项拒绝和截止快照三种合同。