Python asyncio.TaskGroup怎样让一组协程一起失败并统一收集异常怎样完成一个可复用的最小示例

用 asyncio.TaskGroup 管理同一业务操作的多个协程：子任务出现非取消异常时，任务组会请求取消尚未完成的同组任务，等待它们结束，再向外抛出异常组。调用方可在 async with 外用 except* 按类型处理错误。“一起失败”是统一结束和传播的约定，不意味着每个任务同时抛错，也不会回滚已经发生的副作用。

TaskGroup 从 Python 3.11 提供，本文示例只使用 3.11 已有能力，不依赖后续版本新增接口。官方文档：https://docs.python.org/3/library/asyncio-task.html。

1. 把同一业务操作的任务放入共同作用域

把一次聚合查询需要的 profile、stock 和 slow 三项操作看作一个整体：只要缺少必要结果，就不交付完整聚合对象。父协程拥有 TaskGroup，任务字典保留具名任务，子任务的生命周期由共同作用域管理。这是结构化并发在业务接口上的一种具体用法。

如果每一项都是独立下载、允许分别成功失败，就不一定适合遇错取消其余任务；可以选择不同的结果收集策略。关键是先定义业务上的共同成败，而不是看到并发就统一套 TaskGroup。

2. 用最小示例观察失败与清理

保存下面代码为 task_bundle.py。它让 stock 报错，slow 在可取消等待中接受取消，finally 负责演示清理。本文未执行程序，所述现象是代码语义说明。

import asyncio


async def load(name, delay, fail=False):
    try:
        # sleep 模拟可取消的异步等待，不执行真实网络请求
        await asyncio.sleep(delay)
        if fail:
            raise ValueError(f"{name}: 数据不可用")
        return f"{name}:ok"
    finally:
        # 真实客户端应在此释放本任务持有的资源
        print(f"cleanup:{name}")


async def run_bundle():
    tasks = {}
    async with asyncio.TaskGroup() as group:
        tasks["profile"] = group.create_task(load("profile", 0.01))
        tasks["stock"] = group.create_task(load("stock", 0.02, True))
        tasks["slow"] = group.create_task(load("slow", 10))
    # 仅在整个任务组成功退出时读取全部结果
    return {name: task.result() for name, task in tasks.items()}


def leaves(error):
    # 递归展开嵌套结构，保留每个叶子异常的类型
    if isinstance(error, BaseExceptionGroup):
        for child in error.exceptions:
            yield from leaves(child)
    else:
        yield error


async def main():
    try:
        result = await run_bundle()
        print(result)
    except* ValueError as errors:
        # 只消费已明确支持处理的业务错误子组
        for error in leaves(errors):
            print(type(error).__name__, str(error))


if __name__ == "__main__":
    asyncio.run(main())

# 使用 Python 3.11 或以上解释器执行独立示例
python3 task_bundle.py

按示例逻辑，stock 的 ValueError 会离开任务组；slow 不必等待完整十秒即可响应取消并执行清理。profile 可能已经成功，它不需要变成失败。输出中的清理先后顺序不应作为程序契约。任务组异常退出时，不会走到汇总 task.result() 的那一行。

取消是协作式请求：协程必须有机会运行到可响应取消的位置。CPU 密集死循环、同步阻塞库或吞掉 CancelledError 的代码，都可能延迟作用域退出。finally 中若执行异步清理，应遵循对应资源库的取消安全约定；清理代码自己抛错也可能加入失败信息。

3. 用 except* 保留异常的类型结构

普通业务异常通常组成 ExceptionGroup；其叶子可能包括 ValueError、RuntimeError 等类型，except* 处理器接收匹配的子组。CancelledError 用于协作取消，finally 清理与它相关，但普通子任务取消不会作为同类业务错误叶子收集进上述异常组。

示例 leaves 递归访问 BaseExceptionGroup.exceptions，适用于任务组内部还有任务组的场景。不要假设 exceptions 第一层永远都是单个业务异常。except* ValueError 仅处理匹配部分，未匹配的 RuntimeError 等异常仍然向外传播；这比捕获后只打印“任务失败”更能保留定位线索。

KeyboardInterrupt 和 SystemExit 有特殊传播规则，不能把它们当作普通业务错误汇总处理。except* 也不能与普通 except 混放在同一个 try 的处理器列表中；需要时增加内外两层 try。

4. 把任务组接入实际调用边界

实际接口应把客户端连接放在合适的外部资源作用域中，确保任务组先结束，再关闭共享客户端。子任务自身持有的响应体、游标等资源仍由子任务释放。TaskGroup 不会自动替每个库完成资源清理。

任务组本身不提供并发数量上限，也不让同步函数自动变为异步。大批量输入应分批或用 Semaphore 约束真实下游并发；整组期限可结合 asyncio.timeout，子调用还应配置客户端期限。取消线程适配层的等待，也不等于强行终止其底层工作线程。

如果对子任务异常先捕获并返回备用值，任务组会将该任务视为正常完成。只有业务明确允许降级时才这样写，并把降级状态纳入结果；否则会掩盖本应触发整组失败的信号。

为什么只收集到一个错误？

第一个错误出现后，其余任务可能及时取消而未产生自身业务异常。任务组汇总实际发生的错误，不保证每个任务都报告一个失败。

可以在块内直接调用 asyncio.create_task 吗？

这样创建的任务不会自动成为该 TaskGroup 的成员。希望受同组管理时，使用 group.create_task。

TaskGroup 与 gather 的区别是什么？

gather 默认传播首个异常时，不会因此自动取消所有其他 awaitable；TaskGroup 更适合共同生命周期和失败时取消同组任务的场景。选择取决于业务失败策略。