asyncio.run()中未捕获ExceptionGroup会递归打印多行traceback,因它继承BaseException而非Exception,需用except ExceptionGroup:或except BaseException:捕获。

asyncio.run() 里直接抛出 ExceptionGroup 会怎样
Python 3.11+ 中,当多个 asyncio.create_task() 并发失败时,asyncio.gather() 默认会把所有异常聚合成 ExceptionGroup。但如果你用 asyncio.run(main()) 启动,而 main() 内部没处理这个 group,Python 就会把它“展开”成多行 traceback —— 看起来像多个未捕获异常,其实只是同一个 ExceptionGroup 实例在回溯中被递归打印。
这不是 bug,是设计:让开发者意识到“不止一个异常同时发生”。但问题在于,你没法用 except Exception: 捕到它 —— 因为 ExceptionGroup 不继承自 Exception(它继承自 BaseException),且默认不被常规 except 匹配。
- 错误写法:
except Exception:→ 完全捕不到ExceptionGroup - 正确写法:
except BaseException:或更精准的except ExceptionGroup:(需from exceptiongroup import ExceptionGroup,或 Python ≥3.11 且未禁用) - 注意:Python 3.11+ 标准库已内置
ExceptionGroup,无需额外安装,但需确认运行环境版本
用 except ExceptionGroup 捕获但不崩溃
最直接的方式是在调用 asyncio.gather() 的地方加一层 try/except ExceptionGroup,然后手动遍历子异常。关键点在于:不能只打印,得决定哪些异常该重试、哪些该丢弃、哪些要聚合上报。
import asyncio
from exceptiongroup import ExceptionGroup # Python <3.11 才需;3.11+ 可省略此行
<p>async def fetch_user(user_id):
if user_id == 2:
raise ValueError("user not found")
if user<em>id == 3:
raise TimeoutError("network timeout")
return f"user</em>{user_id}"</p><p>async def main():
tasks = [fetch_user(i) for i in [1, 2, 3]]
try:
results = await asyncio.gather(*tasks, return_exceptions=False)
except ExceptionGroup as eg:</p><h1>遍历每个子异常,按类型分别处理</h1><pre class="brush:php;toolbar:false;"> for sub_exc in eg.exceptions:
if isinstance(sub_exc, ValueError):
print(f"→ 忽略用户缺失: {sub_exc}")
elif isinstance(sub_exc, TimeoutError):
print(f"→ 重试超时任务: {sub_exc}")
# 这里可 await fetch_user(3) 再试一次
else:
raise sub_exc # 其他异常仍向上抛
# 注意:此处不能直接 return,因为部分任务可能已成功
# 通常你需要重新 gather 剩余/重试后的任务
立即学习“Python免费学习笔记(深入)”;
-
return_exceptions=False(默认)才能触发ExceptionGroup抛出;设为True则返回含异常对象的列表,不会抛 group -
ExceptionGroup.exceptions是普通 tuple,可遍历、可切片、可filter - 别在
except ExceptionGroup:块末尾静默结束 —— 除非你明确知道所有子异常都已妥善处理
asyncio.wait() + done 集合中手动检查异常
如果你需要更细粒度控制(比如某些任务失败不影响整体流程,或想等部分完成再响应),asyncio.wait() 比 gather() 更合适。它返回 done 和 pending 两个 set,每个 Task 的结果需手动 task.result() 获取,此时异常会以原样抛出(非 group)。
- 好处:每个 task 的异常独立,不会被自动聚合,避免了
ExceptionGroup处理逻辑 - 代价:代码变长,需自己管理 task 生命周期和结果收集
- 典型误操作:对
done中的 task 直接result()而不try/except→ 一旦有异常就中断后续处理
推荐模式:
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
for task in done:
try:
result = task.result() # 成功则取值
print("success:", result)
except Exception as e: # 此处 e 是单个异常,不是 group
print("task failed:", type(e).__name__, e)
日志和监控中怎么识别 ExceptionGroup
很多日志库(如 structlog、loguru)默认不展开 ExceptionGroup,导致报警里只显示 “ExceptionGroup: 2 exceptions” 而看不到具体是什么错。调试时极易漏掉关键子异常。
- 解决方案:在 logger 的
exc_info处理逻辑中,显式调用eg.split()或递归遍历eg.exceptions - 监控系统(如 Sentry)从 v23.x 开始支持
ExceptionGroup自动展开,但需确认 SDK 版本,并开启enable_grouped_exceptions=True - 本地调试时,别只看最后一行 —— 用
traceback.print_exception(eg)替代print(eg),否则只会输出 summary
真正麻烦的不是捕获,而是后续动作:你得决定这个 group 是代表“整体失败”,还是“局部可恢复”。多数业务逻辑里,它其实是后者 —— 但没人帮你做这个判断。


















