Python TaskGroup 如何汇总多个子任务异常

Python 的 asyncio.TaskGroup 会在异步上下文退出时，把多个子任务产生的非取消异常组合成 ExceptionGroup（必要时是 BaseExceptionGroup）再抛出。处理时不要只写一个宽泛的 except Exception；更实用的方式是用 except* 按异常类型拆分子组，让校验错误、超时和其他故障分别进入对应分支。

TaskGroup 改变的是失败边界

我在把一组裸 create_task() 改成 TaskGroup 时，最不适应的不是语法，而是失败不再属于某一个孤立任务。TaskGroup 把同一组子任务放进统一生命周期：其中一个任务首次以非取消异常失败时，组会取消尚未结束的兄弟任务，等待它们退出，然后在 async with 边界统一报告异常。

这项能力从 Python 3.11 加入，核心价值是结构化并发：任务的创建、等待、取消和失败都被限制在清晰的上下文里。对旧代码的直接影响是，原来只接收一个异常的处理逻辑，迁移后可能面对一个有层级的异常组。

最小写法：让 TaskGroup 在出口汇总异常

下面示例一次创建三个任务。子协程中的异常没有在内部吞掉，因此 TaskGroup 退出时会把已经发生的异常放进异常组。示例故意让协程立即失败，便于集中演示异常组；在真实 I/O 中，第一个失败通常会让还未失败的任务收到取消。

import asyncio


async def load_part(name: str) -> str:
    """模拟不同分片的读取失败。"""
    if name.startswith("bad-"):
        # 数据错误保留为 ValueError，便于 except* 分类匹配。
        raise ValueError(f"{name} 数据格式错误")
    # 超时使用独立异常类型，避免与数据错误混在一起。
    raise TimeoutError(f"{name} 读取超时")


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            # 所有任务归属同一个结构化并发边界。
            group.create_task(load_part("bad-orders"), name="orders")
            group.create_task(load_part("bad-users"), name="users")
            group.create_task(load_part("inventory"), name="inventory")
    except* ValueError as group_error:
        # group_error 仍是异常组，只包含匹配的 ValueError 子组。
        for error in group_error.exceptions:
            print(f"数据错误：{error}")
    except* TimeoutError as group_error:
        # 不同类型可以在同一个 try 后分别处理。
        for error in group_error.exceptions:
            print(f"超时错误：{error}")


asyncio.run(main())

except* 后得到的对象不是单个 ValueError，而是保留原有层级的匹配子组。这里直接遍历 exceptions 适合扁平示例；如果子任务内部又创建 TaskGroup，异常组可能嵌套，日志系统应保留组结构和 traceback。

用 except* 按类型拆分异常组

except* 的价值在于“按成员类型匹配”，而不是把整个异常组粗暴归为一种错误。一个异常组可以同时被多个 except* 分支处理：匹配的子组交给当前分支，剩余部分继续匹配后面的分支；最终仍未处理的异常会重新组合并向外传播。

这比在一个 except ExceptionGroup 里手动判断每个成员更稳妥，也能保留嵌套组的上下文。需要注意，同一个 try 不能混用普通 except 和 except*；而且 except* 必须写明匹配类型。

为什么有时只看到一个异常

“汇总多个异常”不等于“等待所有任务都自然失败”。TaskGroup 的安全策略是尽快结束同组工作：第一个普通异常出现后，未完成的兄弟任务会被取消。因此，最终异常组里有几个成员取决于真实并发时序：

这也是我排查“为什么异常组只有一项”时最先看的地方：先确认其他任务究竟已经失败，还是在首个错误后被取消。不能为了凑齐错误列表而吞掉取消；那会破坏 TaskGroup 的结构化并发语义。

取消处理不要吞掉 CancelledError

协程在取消时应通过 finally 释放资源。如果确实捕获 asyncio.CancelledError 做记录，清理后通常要继续 raise。官方文档明确提醒，TaskGroup 和 asyncio.timeout() 都在内部使用取消，吞掉取消可能让这些组件出现异常行为。

import asyncio


async def write_batch() -> None:
    resource = await open_async_resource()
    try:
        # 真实写入逻辑可以在这里包含多个 await。
        await resource.write()
    except asyncio.CancelledError:
        # 可以记录取消，但随后继续抛出，保留 TaskGroup 语义。
        await resource.mark_cancelled()
        raise
    finally:
        # 无论成功、失败还是取消都释放资源。
        await resource.close()

KeyboardInterrupt 和 SystemExit 还有特殊规则：TaskGroup 仍会取消并等待其他任务，但随后重新抛出原始基础异常，而不是把它们当作普通 ExceptionGroup 交给业务分支处理。

TaskGroup 与 gather 应该怎么选

asyncio.gather(..., return_exceptions=True) 也能把异常作为结果返回，但它表达的是“收集每个位置的结果或异常”。TaskGroup 表达的是“这些任务属于同一个生命周期，任一关键失败应让整组收尾”。两者的差别不只是返回值形状。

对我来说，TaskGroup 更适合“一批工作必须整体收口”的场景，例如并行加载同一请求所需的多个数据源；gather(return_exceptions=True) 更适合批量探测、独立检查这类允许部分失败的任务。

给每个异常补上任务上下文

异常组解决了“不要丢异常”，但不会自动把业务对象名称写进错误消息。生产代码最好在子协程边界补充上下文，并用 raise ... from exc 保留原始因果链。这样即使同一类型出现多次，也能知道是哪个数据源、文件或请求失败。

class PartLoadError(RuntimeError):
    """携带分片名称的业务异常。"""


async def load_with_context(part: str) -> bytes:
    try:
        # 调用真正的数据读取函数。
        return await fetch_part(part)
    except OSError as exc:
        # 增加业务标识，同时保留底层异常链。
        raise PartLoadError(f"分片 {part} 读取失败") from exc

迁移旧代码时，我建议先做一个最小验证：准备两个会立即失败的协程和一个会等待的协程，确认异常类型分流、兄弟任务取消、资源清理和未匹配异常传播都符合预期，再把 TaskGroup 放进真实请求链路。

相关问题

TaskGroup 会保证收集到每个子任务的错误吗？

不会。首个非取消异常会触发兄弟任务取消；只有已经发生或在收尾中产生的非取消异常才会进入异常组。

能否直接用 except Exception 捕获 ExceptionGroup？

ExceptionGroup 是 Exception 的子类，普通捕获可以拿到整组，但无法像 except* 那样按成员类型自动拆分。

except* 后拿到的是单个异常吗？

不是。处理器拿到的仍是保持层级的匹配异常子组，可以读取 exceptions 或保留整个组交给日志系统。

官方资料在哪里？

可查看 Python TaskGroup 官方文档、异常组文档和 except* 语法说明。