Python concurrent.futures.as_completed 怎样保留任务异常

用 concurrent.futures.as_completed 读取并发任务时，异常不会在 submit() 那一刻抛出，而是在你对已经完成的 Future 调用 result() 时重新抛出。要保留异常，关键是同时保存“Future 对应哪个任务”的映射，并在循环内部逐个调用 result()。

最小的处理模型是：as_completed() 只负责按完成顺序交出 Future，result() 负责把成功值或原始异常带回当前线程。不要把二者混成一个“批量异常”对象。

官方文档：https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.as_completed

先给 Future 建立任务映射

如果只把 Future 放进列表，异常处理时只能知道“某个任务失败了”，却不知道它对应哪一条输入。更稳妥的做法是用 Future 作为键，把任务编号、文件名或请求参数保存为值。Future 对象可哈希，因此适合建立这种反向索引。

from concurrent.futures import ThreadPoolExecutor, as_completed


def load_record(task_id: str) -> dict:
    # 这里模拟一个可能失败的业务任务；真实项目中可以替换成请求或文件读取。
    if task_id == "bad-2":
        raise ValueError("记录格式不正确")
    return {"task_id": task_id, "value": len(task_id)}


task_ids = ["ok-1", "bad-2", "ok-3"]
with ThreadPoolExecutor(max_workers=3) as executor:
    # 保存 Future 到任务标识的关系，后面捕获异常时不会丢失上下文。
    future_to_task = {
        executor.submit(load_record, task_id): task_id
        for task_id in task_ids
    }

    for future in as_completed(future_to_task):
        task_id = future_to_task[future]
        try:
            # result() 会返回成功值，也会在当前线程重新抛出工作函数的异常。
            record = future.result()
        except Exception as exc:
            print(f"任务 {task_id} 失败：{exc!r}")
        else:
            print(f"任务 {task_id} 完成：{record}")

这段代码有两个边界值得记住：submit() 通常只是提交任务，工作函数里的 ValueError 不会在那里直接抛出；而 future.result() 会把工作线程或进程中的异常重新抛到消费结果的线程。捕获异常的位置应当靠近 result()。

在 result() 处分开成功和失败

as_completed() 产出的是“已经完成或已经取消”的 Future，不代表这些 Future 都成功。一个 Future 完成后仍可能携带异常，所以要把成功值放到 else 分支，把失败信息放到统一的结果结构中。

from concurrent.futures import ThreadPoolExecutor, as_completed


def parse_number(text: str) -> int:
    # 业务函数只负责转换；异常交给外层按任务维度记录。
    return int(text)


inputs = {"a": "12", "b": "not-number", "c": "7"}
results = []

with ThreadPoolExecutor(max_workers=3) as executor:
    future_to_key = {
        executor.submit(parse_number, value): key
        for key, value in inputs.items()
    }

    for future in as_completed(future_to_key):
        key = future_to_key[future]
        try:
            value = future.result()
        except Exception as exc:
            # 保留异常类型和对象，日志或重试层仍可使用原始信息。
            results.append({
                "key": key,
                "ok": False,
                "error_type": type(exc).__name__,
                "error": exc,
            })
        else:
            results.append({"key": key, "ok": True, "value": value})

print(results)

这里的 results 按完成顺序追加，所以顺序不一定和输入字典一致。如果调用方需要输入顺序，应再按 key 排序，或者先给每条任务保留序号。异常对象本身不一定适合直接序列化，跨进程传输或写入 JSON 时可以只保留类型、字符串和任务参数。

取消状态不要当成普通业务异常

Future 可能在任务开始前被取消。对取消的 Future 调用 result() 会抛出 CancelledError，它表达的是调度状态，而不是工作函数执行后返回的业务失败。可以单独捕获它，让重试、统计和日志的含义更清楚。

from concurrent.futures import CancelledError, ThreadPoolExecutor, as_completed


def work(number: int) -> int:
    # 这里保持任务函数简单，重点观察 Future 的消费状态。
    return number * 2


with ThreadPoolExecutor(max_workers=1) as executor:
    futures = [executor.submit(work, number) for number in range(3)]
    # 只有尚未开始运行的任务才可能被取消；返回值表示取消是否成功。
    futures[-1].cancel()

    for future in as_completed(futures):
        try:
            print(future.result())
        except CancelledError:
            # 取消不是工作函数抛出的业务异常，单独统计更准确。
            print("任务在开始前被取消")
        except Exception as exc:
            # 这里处理真正由任务函数产生的异常。
            print(f"任务失败：{exc!r}")

不要只检查 future.cancelled() 就跳过所有状态处理：实际消费逻辑仍应覆盖成功、取消和异常三类结果。若系统把取消视为正常关闭，可以记录为 status="cancelled"；若取消意味着需要补偿，则交给独立的重试策略。

区分单任务异常和整体超时

as_completed(fs, timeout=秒数) 的超时作用在整个迭代器上。它不是某个 Future 的单独超时：从调用 as_completed() 开始计时，如果下次调用 __next__() 时仍没有新的完成项，就会抛出 TimeoutError。因此，单任务异常应在循环内处理，整体超时应在循环外处理。

from concurrent.futures import TimeoutError, ThreadPoolExecutor, as_completed
import time


def slow_work(number: int) -> int:
    # 模拟一个可能超过总等待窗口的任务。
    time.sleep(0.4)
    return number


with ThreadPoolExecutor(max_workers=2) as executor:
    future_to_number = {
        executor.submit(slow_work, number): number
        for number in range(4)
    }
    try:
        # timeout 是整个完成迭代的等待窗口，不是每个 Future 的独立窗口。
        for future in as_completed(future_to_number, timeout=0.5):
            number = future_to_number[future]
            try:
                print("完成", number, future.result())
            except Exception as exc:
                # 只记录已经完成但执行失败的任务。
                print("任务异常", number, repr(exc))
    except TimeoutError:
        # 迭代窗口结束时，仍未完成的 Future 需要另行决定取消或继续等待。
        unfinished = [
            number for future, number in future_to_number.items()
            if not future.done()
        ]
        print("整体等待超时，未完成任务：", unfinished)

捕获外层 TimeoutError 后，已有完成项不会自动消失；未完成项也不会因为迭代器超时而自动取消。可以选择调用 cancel()、记录待处理队列，或把剩余 Future 交给后续的收尾逻辑。是否能取消，取决于任务是否已经开始运行。

把异常保留成可重试的结果

如果失败任务需要重试，不建议在 except 里立即丢掉异常或只打印一行字符串。更实用的结构至少包含任务标识、输入摘要、异常类型、错误文本和重试次数。这样上层可以按异常类型决定是否重试，也可以在最终报告中列出所有失败任务。

from concurrent.futures import ThreadPoolExecutor, as_completed


def run_once(task: tuple[str, int]) -> int:
    # 用 task_id 区分业务对象；真实任务中可以在这里调用外部服务。
    task_id, value = task
    if value  tuple[list[dict], list[dict]]:
    completed, failed = [], []
    with ThreadPoolExecutor(max_workers=4) as executor:
        future_to_task = {
            executor.submit(run_once, task): task
            for task in tasks
        }
        for future in as_completed(future_to_task):
            task_id, value = future_to_task[future]
            try:
                result = future.result()
            except Exception as exc:
                # 失败结果保留原始异常，重试层可按类型做决策。
                failed.append({
                    "task_id": task_id,
                    "input": value,
                    "exception": exc,
                    "retryable": isinstance(exc, TimeoutError),
                })
            else:
                completed.append({"task_id": task_id, "result": result})
    return completed, failed

上例中的 TimeoutError 只是展示按异常类型分类的写法；生产代码应明确区分标准库超时、远端服务异常和参数错误，不能把所有异常都标成可重试。若异常对象最终要发送到 JSON 接口，应在边界层转换为 type、message 等字符串字段，同时保留原异常用于日志。

常见误区和速查表

结论

要在 as_completed 中保留任务异常，只需要抓住三个动作：提交时建立 Future 到任务标识的映射，完成时调用 future.result()，异常时保存原始异常和任务上下文。再把 CancelledError 与迭代器级 TimeoutError 分开处理，日志、重试和最终汇总就不会把不同原因混在一起。

相关问题

• as_completed 会按照提交顺序返回结果吗？不会，它按照 Future 完成的先后顺序返回；需要输入顺序时应自行按任务键重排。

• Future.result() 的异常会丢失原始类型吗？正常情况下会在消费方重新抛出原异常类型；跨进程或跨序列化边界时，应额外设计可传输的错误字段。

• as_completed 超时后未完成任务会自动取消吗？不会。超时只结束本次完成迭代，是否取消要由调用方根据任务状态和业务语义决定。

• 什么时候使用 wait 而不是 as_completed？如果只关心一批任务是否达到某个完成条件，wait() 的集合结果更直接；如果要逐个消费完成结果并保留每条任务异常，as_completed() 更顺手。