Python TaskGroup 怎样让子任务失败统一收敛

如果一组异步任务属于同一个业务操作，通常不应该让调用方逐个等待、逐个猜测失败原因。asyncio.TaskGroup 的核心做法是：一个子任务先抛出非 CancelledError 异常时，取消同组其他任务，等待它们完成清理，然后在 async with 退出处把仍然需要报告的异常集中成 ExceptionGroup。

官方地址：https://docs.python.org/3/library/asyncio-task.html#task-groups

这意味着调用方只需要在任务组边界处理一次失败；子任务则要遵守取消协议，在 finally 中释放连接、文件或临时状态，并在清理完成后继续传播取消。

TaskGroup 的失败会怎样传播

TaskGroup 适合表达“这些任务要一起成功，任意一个关键任务失败就结束这一组”的关系。它不是把所有任务都跑完再收集结果：首个非取消异常会成为收敛信号，剩余任务会收到取消请求。

例如“读取配置、库存和价格”属于同一个页面响应，库存请求失败后，继续等待价格请求就没有意义。TaskGroup 会先让价格任务收尾，再把库存任务的异常带到上下文外。这样可以避免后台任务在主流程已经失败后继续占用连接。

先写出正确的取消收尾

取消不是普通业务异常。子任务在下一个可中断的 await 处收到 CancelledError，应使用 finally 做清理；如果显式捕获它，清理结束后通常还要重新抛出。吞掉取消会让 TaskGroup 的退出过程失去可靠边界。

import asyncio


async def load_part(name: str, delay: float, should_fail: bool = False) -> str:
    resource = f"连接:{name}"
    try:
        # 模拟一次可被取消的异步 I/O。
        await asyncio.sleep(delay)
        if should_fail:
            raise ValueError(f"{name} 返回业务错误")
        return f"{name} 已完成"
    except asyncio.CancelledError:
        # 记录取消或撤销临时状态，清理后继续传播取消。
        raise
    finally:
        # 真实代码在这里关闭连接、释放锁或删除临时文件。
        resource = None


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(load_part("配置", 0.2))
            group.create_task(load_part("库存", 0.1, should_fail=True))
            group.create_task(load_part("价格", 2.0))
    except* ValueError as errors:
        # 只处理业务失败；同组被取消的任务不会作为普通错误上报。
        for error in errors.exceptions:
            print(f"需要重试或降级: {error}")


asyncio.run(main())

这段代码的关键不在于打印错误，而在于 load_part 没有把取消当成成功返回。finally 无论任务成功、失败还是被取消都会执行；except asyncio.CancelledError 只负责完成取消前的必要动作，然后用 raise 保留取消状态。

用 except* 拆分统一收敛的异常

TaskGroup 退出时，多个非取消异常会被组合进 ExceptionGroup。Python 的 except* 可以按异常类型匹配其中一部分，让业务异常、超时异常和未知异常分别进入不同处理分支。它和普通 except 的区别是：处理的是异常组中的匹配成员。

import asyncio


async def call_service(name: str) -> None:
    # 每个子任务代表一个可以独立失败的外部调用。
    await asyncio.sleep(0)
    if name == "库存":
        raise ValueError("库存数据格式不正确")
    if name == "价格":
        raise TimeoutError("价格服务响应超时")


async def run_batch() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(call_service("配置"))
            group.create_task(call_service("库存"))
            group.create_task(call_service("价格"))
    except* ValueError as errors:
        # 业务数据错误可以转为降级结果或告警。
        for error in errors.exceptions:
            print(f"业务失败: {error}")
    except* TimeoutError as errors:
        # 超时通常进入重试、兜底缓存或快速失败策略。
        for error in errors.exceptions:
            print(f"超时失败: {error}")


asyncio.run(run_batch())

实际运行时，如果一个任务先失败，TaskGroup 可能取消尚未执行到抛错位置的其他任务，因此不应把“每个分支一定都会产生一个异常”写进业务假设。处理逻辑应该面向最终收到的异常类型，并为未匹配的异常保留向上抛出的路径。

取消、业务失败和进程级异常要分开

尤其不要在子任务里写一个宽泛的 except Exception 后把所有情况转换成“失败结果”。CancelledError 直接继承自 BaseException，通常不会被这个分支捕获；更重要的是，统一吞错会让调用方无法区分用户主动取消、业务错误和系统退出。

把超时放在任务组的边界上

如果整组任务有一个总时限，可以把 TaskGroup 放进 asyncio.timeout()。超时发生后，任务组内部的子任务会收到取消，外层再根据超时异常做降级。子任务仍然必须在 finally 中释放自己的资源。

import asyncio


async def load_dashboard() -> None:
    try:
        # timeout 约束整组工作，不只约束某一次 sleep。
        async with asyncio.timeout(1.0):
            async with asyncio.TaskGroup() as group:
                group.create_task(load_part("配置", 0.2))
                group.create_task(load_part("库存", 0.8))
                group.create_task(load_part("价格", 0.9))
    except TimeoutError:
        # 总时限耗尽后返回缓存、降级数据或明确的超时结果。
        print("整组任务超时，进入降级路径")


asyncio.run(load_dashboard())

边界要注意两点：第一，超时是外层控制流的结果，不应在每个子任务里重复把取消改写成普通错误；第二，finally 里的清理要短小且可中断，避免清理逻辑本身把 TaskGroup 的退出拖成新的问题。

排查 TaskGroup 没有按预期收敛

• 确认任务是在 async with TaskGroup() 进入后通过 create_task() 加入的，而不是提前创建后遗留在外部。
• 检查子任务是否捕获了 CancelledError 却没有重新抛出，或在取消后继续执行长时间不可中断的工作。
• 检查调用方是否只处理了一个异常类型，导致异常组中未匹配的成员继续向上抛出。
• 确认超时的位置：总时限放在任务组外层，单个服务的时限才放在对应子任务内部。
• 不要把静态说明图或日志中的“取消”当成成功；真正的成功应由任务返回值和业务结果决定。

常见问题

TaskGroup 会等待所有子任务自然结束吗？

不一定。没有异常时，它会等待所有任务完成；出现首个非取消异常后，会取消同组剩余任务并等待它们完成取消收尾。

为什么没有在 ExceptionGroup 里看到 CancelledError？

TaskGroup 会把正常的任务取消视为收尾信号，而不是需要集中报告的业务失败。真正需要处理的是非取消异常。

可以用普通 except 捕获 TaskGroup 的异常吗？

可以捕获整个 ExceptionGroup，但如果要按子异常类型分别处理，更适合使用 except*。未被处理的成员仍会继续传播。

子任务里应该返回错误对象而不是抛异常吗？

只有当“单个失败也允许整组成功”是明确业务规则时才返回错误对象。若任一任务失败就应终止这次操作，直接抛出异常才能让 TaskGroup 的结构化取消发挥作用。

总结一下：用 TaskGroup 表达同一业务边界，用子任务的 finally 负责可重复的清理，用 except* 在边界一次性拆分失败。只要不吞掉取消信号，子任务失败就能从“各自散落的 await”收敛为可处理的异常组。