asyncio.gather默认将所有异常聚合成ExceptionGroup,即使仅一个协程出错;需用except*捕获特定类型异常,或设return_exceptions=True避免ExceptionGroup。

asyncio.gather 默认把多个异常打包成 ExceptionGroup
Python 3.11 引入 ExceptionGroup 后,asyncio.gather 在多个协程出错时不再抛出第一个异常,而是把所有异常聚合成一个 ExceptionGroup。这意味着你不能用老办法——比如只捕获 ValueError 或 Exception——来拿到全部错误信息。
常见错误现象:
你写了 await asyncio.gather(a(), b(), c()),其中 b() 和 c() 都抛了 TimeoutError,但 try-except 块里什么也没捕获到,或者只看到 ExceptionGroup: 2 exceptions 而无法访问具体异常。
-
asyncio.gather(..., return_exceptions=False)(默认):任一子协程失败 → 整体 raiseExceptionGroup -
asyncio.gather(..., return_exceptions=True):所有异常都转为结果列表里的Exception实例,不触发ExceptionGroup - 即使只一个协程出错,也会包装成含单个异常的
ExceptionGroup(不是原异常类型)
用 except* 捕获 ExceptionGroup 中的特定异常类型
except* 是 Python 3.11 新增语法,专为 ExceptionGroup 设计。它会从组中“筛选”出匹配类型的异常子集,而不是要求整个组匹配某个类型。
示例场景:你想分别处理 TimeoutError 和 ConnectionError,且不希望漏掉任何一个:
立即学习“Python免费学习笔记(深入)”;
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) - 不能混用
except和except*在同一个 try 块里(SyntaxError)
手动解包 ExceptionGroup 获取原始异常链
有时候你需要检查异常上下文、traceback 或触发原因(比如哪个 task 抛了错),而 except* 只给子集。这时得手动遍历 exceptions 属性,并注意嵌套可能:
-
ExceptionGroup.exceptions是元组,元素可能是普通异常,也可能是另一层ExceptionGroup - 推荐用
exceptiongroup.ExceptionGroup.split()辅助分类(需安装exceptiongroupbackport 包用于 3.11 之前,但 3.11+ 已内置) - 若要保留 traceback,别用
str(e),改用traceback.format_exception(type(e), e, e.__traceback__)
小技巧:在日志中打印完整结构:
import tracebacktry: 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=True→gather总是返回 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”。










