Python asyncio.TaskGroup 中一个任务失败后如何安全收集结果：异常聚合与取消传播

批量请求三个下游服务时，最麻烦的不是某个请求失败，而是失败之后其他任务到底处于什么状态。asyncio.TaskGroup 的默认语义很明确：同组任务第一次抛出非 CancelledError 异常后，剩余任务会收到取消；所有任务结束后，异常会以 ExceptionGroup 的形式从上下文管理器抛出。

想安全收集结果，就把“成功结果”和“失败异常”分开保存，在 TaskGroup 外部用 except* 拆异常；不要在子协程里吞掉 CancelledError，否则取消收敛会变得不可预测。

TaskGroup 解决的是哪一段并发风险

下面用三个独立的下游任务模拟批量聚合。fetch_profile 成功，fetch_orders 主动失败，fetch_quota 可能正在等待。调用方真正关心的是：失败是否会取消同组任务、异常能否被统一捕获，以及成功结果有没有被误当成完整结果。

import asyncio

async def fetch_profile():
    await asyncio.sleep(0.05)
    return {"name": "Lin"}

async def fetch_orders():
    await asyncio.sleep(0.02)
    raise RuntimeError("orders backend unavailable")

async def fetch_quota():
    try:
        await asyncio.sleep(1)
        return {"left": 8}
    finally:
        print("fetch_quota cleanup")

这里的三个正文节点是 fetch_profile、fetch_orders 和 fetch_quota。它们不是装饰性的函数名，而是后面两张图要核对的真实调用链。

第一个异常出现后，取消是怎样传播的

把任务保存下来，再让 async with 自然退出。TaskGroup 会等待被取消的 fetch_quota 执行完 finally 清理，然后把失败信息交给外层。父协程在上下文内部可能被内部取消唤醒，但这个取消不会直接穿出 async with。

async def load_dashboard():
    async with asyncio.TaskGroup() as group:
        profile_task = group.create_task(fetch_profile())
        orders_task = group.create_task(fetch_orders())
        quota_task = group.create_task(fetch_quota())

    return {
        "profile": profile_task.result(),
        "orders": orders_task.result(),
        "quota": quota_task.result(),
    }

async def main():
    try:
        await load_dashboard()
    except* RuntimeError as errors:
        for error in errors.exceptions:
            print("grouped:", error)

asyncio.run(main())

运行时通常会先看到 fetch_quota cleanup，随后才看到分组异常。这里不要在 load_dashboard 中用普通的 except Exception 试图把失败任务变成空字典：那会掩盖“页面数据不完整”这个业务状态。

结果收集为什么不能直接遍历所有 task

TaskGroup 退出后，成功任务可以通过 result() 读取；失败任务的 result() 会重新抛出异常，被取消的任务也不会产生业务结果。因此更稳妥的接口是返回“部分结果 + 分组异常”，而不是把失败伪装成成功响应。

async def load_partial():
    tasks = {}
    try:
        async with asyncio.TaskGroup() as group:
            tasks["profile"] = group.create_task(fetch_profile())
            tasks["orders"] = group.create_task(fetch_orders())
            tasks["quota"] = group.create_task(fetch_quota())
    except* RuntimeError as errors:
        failed = [str(error) for error in errors.exceptions]
        ok = {}
        for name, task in tasks.items():
            if not task.cancelled() and not task.exception():
                ok[name] = task.result()
        return ok, failed
    return {name: task.result() for name, task in tasks.items()}, []

如果业务要求“三个数据齐全才渲染”，调用方应该检查失败列表，而不是只看 ok 字典非空。部分成功可以用于降级展示，但必须显式标记数据不完整。

CancelledError 的清理边界

取消不是普通业务异常。官方文档把 CancelledError 设计为 BaseException 的子类，协程可以在 finally 中关闭连接、删除临时文件或记录状态；清理结束后通常应继续抛出它。若确实要抑制取消，还要同步处理任务的取消状态，不能只写一个空的 except。

async def fetch_quota():
    resource = await open_resource()
    try:
        return await resource.read()
    finally:
        await resource.close()

实际项目中还要检查清理动作本身是否可取消，必要时把幂等的收尾写成短路径。不要在清理阶段继续启动新的后台任务，也不要把取消转换成“配额为空”。

三个容易误判的验收点

把 ExceptionGroup 当成单个异常

使用 except* RuntimeError 才能按异常类型处理分组内容；普通 except RuntimeError 不会直接匹配外层的 ExceptionGroup。

在 TaskGroup 内读取失败任务结果

上下文管理器尚未退出时，任务可能还在收敛。读取结果应放在组外，并对 cancelled() 与 exception() 做判断。

吞掉 CancelledError 让任务“继续完成”

这样会破坏结构化并发的退出语义，尤其是在超时、嵌套 TaskGroup 或外部取消同时发生时。清理完成后重新抛出是更安全的默认选择。

把这套模式放进真实聚合接口

建议把聚合函数的返回值固定为“数据、失败原因、是否完整”三部分。HTTP 层可以据此选择完整响应、降级响应或重试，而不是靠捕获日志猜测哪个下游被取消。

验收时至少覆盖三种场景：所有任务成功；一个任务抛出 RuntimeError 且另一个任务进入清理；调用方在任务组外部被取消。只要能观察到清理发生、异常可分组、部分结果不伪装成完整结果，这个并发边界才算真正落地。

相关问题

TaskGroup 和 gather 应该怎么选？

需要同组生命周期和失败即取消时优先考虑 TaskGroup；需要保留每个任务的异常结果并自行决定是否取消时，再评估 gather 的返回策略。

为什么捕获 CancelledError 后程序仍然退出？

因为取消是协作式的，任务在下一个可中断点收到它。捕获只适合做清理，清理完成后继续抛出才能让上层知道任务确实被取消。

小结

TaskGroup 的价值不在于让错误消失，而在于把任务的创建、等待、取消和异常汇总放到同一条生命周期里。写聚合接口时，保留成功结果、明确失败集合，并尊重 CancelledError 的传播规则，才能让降级和重试建立在真实状态上。