Python 3.11+ 的 ExceptionGroup 如何与 asyncio.gather 配合捕获

技术百科 冷炫風刃 发布时间:2026-01-27 浏览:
asyncio.gather默认将所有异常聚合成ExceptionGroup,即使仅一个协程出错;需用except*捕获特定类型异常,或设return_exceptions=True避免ExceptionGroup。

asyncio.gather 默认把多个异常打包成 ExceptionGroup

Python 3.11 引入 ExceptionGroup 后,asyncio.gather 在多个协程出错时不再抛出第一个异常,而是把所有异常聚合成一个 ExceptionGroup。这意味着你不能用老办法——比如只捕获 ValueErrorException——来拿到全部错误信息。

常见错误现象:
你写了 await asyncio.gather(a(), b(), c()),其中 b()c() 都抛了 TimeoutError,但 try-except 块里什么也没捕获到,或者只看到 ExceptionGroup: 2 exceptions 而无法访问具体异常。

  • asyncio.gather(..., return_exceptions=False)(默认):任一子协程失败 → 整体 raise ExceptionGroup
  • asyncio.gather(..., return_exceptions=True):所有异常都转为结果列表里的 Exception 实例,不触发 ExceptionGroup
  • 即使只一个协程出错,也会包装成含单个异常的 ExceptionGroup(不是原异常类型)

用 except* 捕获 ExceptionGroup 中的特定异常类型

except* 是 Python 3.11 新增语法,专为 ExceptionGroup 设计。它会从组中“筛选”出匹配类型的异常子集,而不是要求整个组匹配某个类型。

示例场景:你想分别处理 TimeoutErrorConnectionError,且不希望漏掉任何一个:

try:
    await asyncio.gather(task_a(), task_b(), task_c())
except* TimeoutError as eg:
    print(f"超时了 {len(eg.exceptions)} 次")
    for e in eg.exceptions:
        print(" -", e)
except* ConnectionError as eg:
    print("连接问题,忽略重试")
except* Exception as eg:
    # 捕获剩下所有未被前面 except* 匹配的异常
    print("其他异常:", eg)
  • except* 不是“或”关系,而是“提取子集”:每个块只看到自己类型的部分异常
  • 多个 except* 块可共存,互不影响;顺序无关(不像普通 except
  • 不能混用 exceptexcept* 在同一个 try 块里(SyntaxError)

手动解包 ExceptionGroup 获取原始异常链

有时候你需要检查异常上下文、traceback 或触发原因(比如哪个 task 抛了错),而 except* 只给子集。这时得手动遍历 exceptions 属性,并注意嵌套可能:

  • ExceptionGroup.exceptions 是元组,元素可能是普通异常,也可能是另一层 ExceptionGroup
  • 推荐用 exceptiongroup.ExceptionGroup.split() 辅助分类(需安装 exceptiongroup backport 包用于 3.11 之前,但 3.11+ 已内置)
  • 若要保留 traceback,别用 str(e),改用 traceback.format_exception(type(e), e, e.__traceback__)

小技巧:在日志中打印完整结构:

import traceback

try: await asyncio.gather(a(), b(), c()) except ExceptionGroup as eg: print("完整异常组:") for i, exc in enumerate(eg.exceptions): print(f"[{i}] {type(exc).name}: {exc}")

打印 traceback(仅限

非嵌套异常)

    if not isinstance(exc, ExceptionGroup):
        tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
        print(tb.strip().split("\n")[-3:])  # 只看最后三行

与 return_exceptions=True 配合时无需 ExceptionGroup 处理

如果你更习惯传统错误处理逻辑(比如逐个检查结果、手动 raise),可以关掉 ExceptionGroup 自动聚合:

  • return_exceptions=Truegather 总是返回 list,成功结果和异常对象并存
  • 此时你得自己遍历结果,用 isinstance(r, Exception) 判断是否出错
  • 优点:兼容旧代码、调试直观;缺点:丢失异常聚合语义(比如无法区分“三个 timeout”和“一个 timeout + 两个 json decode error”)

示例:

results = await asyncio.gather(
    fetch("a"), fetch("b"), fetch("c"),
    return_exceptions=True
)
for i, r in enumerate(results):
    if isinstance(r, Exception):
        print(f"task {i} failed: {r}")
        # 这里可以单独处理,比如记录、重试、raise 等

真正容易被忽略的是:哪怕你只启动了一个协程,asyncio.gather(single_coro()) 出错时依然返回 ExceptionGroup——它不是“多任务专属”,而是 gather 的统一错误模型。别假设“只有一个任务就不会遇到 ExceptionGroup”。


# ai  # 的是  # 写了  # 不像  # 多个  # 第一个  # 你想  # 也会  # python  # 你不  # js  # json  # Error  # 对象  # 重试  # try  # 遍历  # raise 


相关栏目: <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 AI推广<?muma echo $count; ?> 】 <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 SEO优化<?muma echo $count; ?> 】 <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 技术百科<?muma echo $count; ?> 】 <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 谷歌推广<?muma echo $count; ?> 】 <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 百度推广<?muma echo $count; ?> 】 <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 网络营销<?muma echo $count; ?> 】 <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 案例网站<?muma echo $count; ?> 】 <?muma $count = M('archives')->where(['typeid'=>$field['id']])->count(); ?> 【 精选文章<?muma echo $count; ?>

相关推荐

在线咨询

点击这里给我发消息QQ客服

在线咨询

免费通话

24h咨询:4006964355


如您有问题,可以咨询我们的24H咨询电话!

免费通话

微信扫一扫

微信联系
返回顶部