Python asyncio TaskGroup 怎样收集多个失败

我们平时用asyncio的TaskGroup管理多个异步任务时，默认行为是只要任意一个子任务抛出异常，就会立刻取消所有正在运行的其他子任务，最终仅向外抛出第一个触发的异常，没办法直接拿到所有任务的错误信息，想要完整收集多个任务的失败，可以通过给每个协程套一层异常捕获、搭配自定义聚合异常类存储所有错误的方案实现。

实际业务里不建议直接修改TaskGroup内部逻辑，优先用协程包装层兜底异常，不会触发TaskGroup的默认取消连锁反应，所有错误都能完整留存下来统一后续处理。

我第一次把多个异步任务改成 asyncio.TaskGroup 时，原本以为退出上下文后可以像读取任务列表一样拿到所有错误。实际情况是：TaskGroup 更像“带安全边界的并发任务组”，首个非取消异常会触发其余任务的取消，退出时再把已经产生的异常组合成 ExceptionGroup。所以“收集多个失败”有两种意思：要分类已经传播出来的异常，用 except*；要统计每个业务任务的失败，就在子任务内部把异常转成结果对象。

需要保留异常语义时，在 TaskGroup 外层捕获 ExceptionGroup；需要每个任务都给出成功或失败结果时，在任务函数内部捕获普通异常，但要继续向上传播 CancelledError。

先分清 TaskGroup 的失败边界

TaskGroup 从 Python 3.11 引入。它会等待组内任务，某个任务首次抛出非 asyncio.CancelledError 异常后，会取消尚未完成的同组任务；所有任务收尾后，剩余的非取消异常会组合为 ExceptionGroup 或 BaseExceptionGroup。这意味着它不是“等所有任务自然失败再逐个读取异常”的容器。

如果你只想知道这次并发操作属于哪类故障，异常组正是合适的出口；如果你必须得到 20 个任务中每一个的状态，就不能只依赖 TaskGroup 的异常传播结果。

用 except* 分类 TaskGroup 传播出的多个异常

当多个子任务在取消发生前都已经抛出异常，TaskGroup 会把它们放进异常组。Python 的 except* 可以按异常类型拆分这组异常；处理器拿到的是匹配后的异常组，使用它的 exceptions 属性可以遍历当前层的成员。

import asyncio

async def fetch_one(name: str) -> None:
    # 示例只表达异常传播边界，不代表真实网络请求。
    await asyncio.sleep(0)
    if name == "库存":
        raise TimeoutError("库存服务超时")
    raise ValueError(f"{name} 返回格式不正确")

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            # TaskGroup 负责持有任务，并在退出时等待收尾。
            group.create_task(fetch_one("库存"))
            group.create_task(fetch_one("价格"))
    except* TimeoutError as errors:
        # except* 只处理匹配到的 TimeoutError 子集。
        for error in errors.exceptions:
            print("可重试错误：", error)
    except* ValueError as errors:
        # 业务数据错误单独记录，避免和超时重试混在一起。
        for error in errors.exceptions:
            print("数据错误：", error)

asyncio.run(main())

这里的两个任务都在第一次让出控制权后抛错，实际异常成员仍由调度时机决定，不能把示例输出数量写死。工程代码应把 except* 当成“按类型处理已经收集到的异常”，不要把它理解成绕过 TaskGroup 的取消策略。

业务要统计全部任务时，把失败变成结果

批量校验、批量抓取或批量生成报告，通常更关心“每个项目成功了吗、失败原因是什么”。这时让每个子任务自己捕获普通异常，返回统一的结果对象，TaskGroup 就能正常等待所有任务；取消仍然单独处理，不能被当成普通业务失败吞掉。

import asyncio
from dataclasses import dataclass

@dataclass
class JobResult:
    name: str
    ok: bool
    error: str | None = None

async def run_job(name: str, should_fail: bool) -> JobResult:
    try:
        # 这里代表一次异步业务调用，真实项目可替换为 HTTP 或数据库操作。
        await asyncio.sleep(0)
        if should_fail:
            raise ValueError(f"{name} 校验失败")
        return JobResult(name, True)
    except asyncio.CancelledError:
        # 取消是结构化并发的控制信号，清理后必须继续向上传播。
        raise
    except Exception as error:
        # 普通业务异常转成数据，保证调用方能看到每个任务的结果。
        return JobResult(name, False, f"{type(error).__name__}: {error}")

async def main() -> None:
    async with asyncio.TaskGroup() as group:
        tasks = [
            group.create_task(run_job("订单-1", False)),
            group.create_task(run_job("订单-2", True)),
            group.create_task(run_job("订单-3", True)),
        ]
    # 离开上下文后，任务已经完成，可以安全读取 result()。
    results = [task.result() for task in tasks]
    failures = [item for item in results if not item.ok]
    print("失败数量：", len(failures))
    for item in failures:
        print(item.name, item.error)

asyncio.run(main())

这种写法收集的是业务结果，不是让 TaskGroup 同时传播多个异常。若某个任务被外部取消，CancelledError 仍会让任务组按取消语义收尾；如果协程把它当成普通异常返回，超时、关闭和上层取消就可能被误报成业务失败。

两个容易踩的边界

第一，不要在普通的 except Exception 里顺手捕获取消信号。CancelledError 直接继承自 BaseException，虽然通常不会被 except Exception 捕获，但显式捕获后也应在清理完成后重新抛出。第二，TaskGroup 的 create_task() 只能在活动的任务组上下文中调用；离开上下文后再添加任务会失败。

如果只是希望并发任务互不影响、每个任务独立拿到异常结果，可以使用结果对象模式；如果希望首个故障触发组内收敛并统一处理，就保留异常传播，用 except* 拆分类型。两者不是写法优劣，而是失败语义不同。

相关问题

TaskGroup 能保证收集到所有任务的异常吗？

不能保证所有任务都来得及抛出并被收集。首个非取消异常会取消尚未完成的任务；要统计每个任务的业务失败，应在任务内部转换成结果。

为什么不能把 CancelledError 当成普通失败返回？

因为取消是 TaskGroup 和上层调用方协作收尾的控制信号。吞掉它可能让结构化并发组件无法正确结束或丢失外部取消。

except* 和 except 有什么区别？

except 面向一个普通异常；except* 可以从 ExceptionGroup 中按类型拆出匹配成员，适合 TaskGroup 的组合异常。