asyncio TaskGroup 怎样收集并发异常

在 async with asyncio.TaskGroup() 外使用 except*，按异常类型处理异常组；需要逐条记录时，再递归遍历组内的异常叶子。TaskGroup 会等待组内任务收尾，并汇总实际发生的非取消异常，但它不会保证所有任务都运行到各自报错的位置。

本文用一个批次检查场景说明怎样统计异常：一项任务报告字段和附件错误，另一项任务等待并响应取消。示例为原创演示，未在本机执行；所有数量说明都是该示例的结构或语义预期，不是压测成绩。

官方文档：https://docs.python.org/3/library/asyncio-task.html#task-groups。异常组说明：https://docs.python.org/3/library/exceptions.html#exception-groups。语法参考：https://docs.python.org/3/reference/compound_stmts.html#except-star。

先定义观察指标：异常数不等于任务数

先把提交任务数、异常叶子数、取消任务数分开。两个任务并不意味着最多只有两个异常：某个任务可能抛出嵌套异常组。另一方面，任务被取消也不应计作一次附件读取失败。本文记录的是分类后的叶子数量，并单独观察 Task.cancelled()。

建立假设：失败会取消同组未完成任务

普通子任务首次发生非 CancelledError 异常时，TaskGroup 会请求取消其他尚未完成的同组任务，等待收尾后抛出异常组。因此，“收集并发异常”的准确含义是收集退出前实际产生的错误，不能理解为强制每个任务完成全部检查。

示例的 batch_check 自身抛出一个包含 ValueError 和 OSError 的 ExceptionGroup；pending_job 则保持等待。ready 是两个协程共享的就绪约定，保证等待任务已经进入其工作范围。这个设计刻意避开用两个相近 sleep 时间来赌谁先抛错。

改动点：在 TaskGroup 外按类型处理异常组

保存以下代码为 taskgroup_errors.py，使用 Python 3.11 或以上版本。except* 捕获到的是匹配类型的异常子组，内部结构仍可能嵌套，所以 leaves 不直接把 errors.exceptions 当成平坦错误清单。

import asyncio


def leaves(error):
    # 异常组可以嵌套，递归后才是实际异常叶子。
    if isinstance(error, BaseExceptionGroup):
        for child in error.exceptions:
            yield from leaves(child)
    else:
        yield error


async def batch_check(ready):
    await ready.wait()
    # 一个子任务主动报告两个检查错误，不冒充两项独立并发失败。
    raise ExceptionGroup("批次检查失败", [
        ValueError("字段 region 无效"),
        OSError("附件读取失败"),
    ])


async def pending_job(ready):
    ready.set()
    try:
        # 保持挂起，便于观察同组失败引起的取消。
        await asyncio.Event().wait()
    finally:
        print("挂起任务清理完成")


async def main():
    ready = asyncio.Event()
    tasks = {}
    counts = {"value": 0, "io": 0}
    try:
        async with asyncio.TaskGroup() as tg:
            tasks["check"] = tg.create_task(batch_check(ready))
            tasks["pending"] = tg.create_task(pending_job(ready))
    except* ValueError as errors:
        # 只消费本例约定的输入错误，保留每个错误的类型和消息。
        for error in leaves(errors):
            counts["value"] += 1
            print(type(error).__name__, str(error))
    except* OSError as errors:
        for error in leaves(errors):
            counts["io"] += 1
            print(type(error).__name__, str(error))
    # 未匹配异常会继续传播，不会伪装成成功。
    print("已处理叶子数：", counts)
    print("取消任务数：", sum(task.cancelled() for task in tasks.values()))


asyncio.run(main())

这里的两个 except* 子句可以在同一次异常处理中分别执行。ValueError 子组交给字段类统计，OSError 子组交给读取类统计；未被匹配的错误继续向外传播。示例打印消息方便理解，实际服务应把记录送入项目日志系统，并明确哪些错误允许被消费。

复现方法：记录叶子异常和取消状态

# 在支持 TaskGroup 和 except* 的 Python 环境执行示例。
python3 taskgroup_errors.py

读者复现时先关注“挂起任务清理完成”，再关注各异常类型和计数。这个输出来自 finally 中的清理说明，并没有把取消转换成成功返回。示例没有持有真实网络连接；若换成实际客户端，资源关闭逻辑应放在相应的异步上下文或 finally 中。

生产排查时还应保留完整 traceback。只保存 str(error) 可能丢失调用位置、异常链和任务来源。可在业务异常中携带任务标识，或在创建任务时维护业务元数据；不要凭异常在组里的位置推断它属于第几个任务。

结果对比：看异常结构，不看虚构耗时

这里两个错误来自同一个批次检查任务，不能据此宣称 TaskGroup 总能同时捕获两个独立故障。如果换成两个各自等待 I/O 的任务，先出现的失败可能使另一个任务取消；实际异常叶子数量由已经发生的错误决定。

这项改动的价值在于异常不再被单条错误消息遮住，并且取消有独立口径。本文没有测量吞吐、内存和耗时，也不把异常处理结构的改动包装为性能提升。若业务真要衡量开销，应在自己的负载里区分成功路径、失败路径和清理成本。

边界条件与常见问题

为什么普通 except ValueError 没有处理到子错误？

外层对象是异常组，按子异常类型处理应使用 except*。同一个 try 语句不能同时混写普通 except 和 except*；如果必须处理不同层级，应使用嵌套 try 并明确各层职责。

能否把取消也放进失败清单？

可以另记取消状态，但不要把它误标为业务异常。CancelledError 属于 BaseException 层级，协程响应取消后通常应继续传播；误吞它可能破坏 TaskGroup 的收尾行为。KeyboardInterrupt 和 SystemExit 还有特殊传播规则，不宜用普通错误统计框架拦截。

想让每条记录都检查完怎么办？

那是不同的业务约定。可在单项任务内部把明确可恢复的校验错误转换为结构化结果，使任务正常返回；同时让未知程序错误继续抛出。不要仅在 TaskGroup 外增加捕获，就期待被取消的检查自动完成。

except* 处理后怎样保留失败状态？

本例为了展示计数，消费了已知类型。在服务入口应按业务决定返回失败对象、记录状态或重新抛出异常；“错误已记录”并不等于整个请求成功。统计、补偿和对调用方反馈应使用同一套业务口径。