asyncio.TaskGroup 更安全,因其在任一子任务异常时自动取消其余任务并等待清理,避免资源泄漏;而 gather 会让所有任务继续执行。

为什么 asyncio.TaskGroup 比 asyncio.gather() 更安全?
因为 TaskGroup 在任意子任务抛出异常时会自动取消其余正在运行的任务,并等待它们清理完毕;而 gather(return_exceptions=False) 会直接让所有任务继续跑完,可能造成资源泄漏或状态不一致。
典型场景:你启动了 3 个异步 HTTP 请求 + 1 个数据库写入,其中第 2 个请求返回 500。用 gather 时,其余请求和写入仍会执行;用 TaskGroup,一旦异常抛出,其他任务收到 asyncio.CancelledError 并有机会在 finally 或异步上下文管理器中释放连接。
- 必须在
async with asyncio.TaskGroup() as tg:块内调用tg.create_task(),不能提前保存 task 对象再 await -
TaskGroup是 Python 3.11+ 原生特性,3.10 及更早需用第三方 backport(如asyncio-taskgroup包) - 若某任务因未捕获异常退出,
TaskGroup会把该异常作为ExceptionGroup的一部分重新抛出,不是单个异常
asyncio.TaskGroup 中如何处理多个异常?
当多个子任务分别抛出不同异常(比如一个超时、一个 JSON 解析失败、一个网络断连),TaskGroup 不会只抛第一个,而是聚合为 ExceptionGroup。你需要显式检查或匹配。
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(fetch_user(1))
tg.create_task(fetch_user(2))
tg.create_task(fetch_user(3))
except ExceptionGroup as eg:
for exc in eg.exceptions:
if isinstance(exc, asyncio.TimeoutError):
print("有请求超时")
elif isinstance(exc, ValueError):
print("有解析错误")
- 不能用
except ValueError:直接捕获子任务中的ValueError,它被包在ExceptionGroup里了 -
eg.subgroup((TypeError, KeyError))可提取特定类型子集,适合做分类处理 - 如果只想忽略某类异常(如部分请求失败可接受),应在子任务内部处理,而不是依赖外层 except
什么时候不该用 TaskGroup?
当你需要精确控制每个任务生命周期、或要求“只要有一个成功就立即返回”时,TaskGroup 就不合适——它坚持“全进全出”,所有任务都必须进入完成/取消状态才退出上下文。
SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能,主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的 双功能合体技能 ,专为学生、算法学习者和开发者设计;✅ 一个技能,两种用途 :;核心要点📝 题解模式 :输入题目/题号,自动生成完整 Python 题解(含详细注释、解题思路、复杂度分析);💬 注释模式 :输入 Python 代码,自动添加详细中。它将相关步骤、
立即学习“Python免费学习笔记(深入)”;
- 竞速场景(race):改用
asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED) - 后台常驻任务(如心跳、日志轮转):应单独用
asyncio.create_task()并持有引用,避免被TaskGroup统一 cancel - 任务之间有强顺序依赖(B 必须等 A 的返回值):这不是并发问题,应串行 await,强行塞进
TaskGroup反而增加复杂度
常见报错:RuntimeError: Task group is closed
这个错误几乎总是因为你试图在 async with 块**之外**调用 tg.create_task(),或者在 __aexit__ 已触发后还尝试提交任务。
典型误写:
tg = None
async with asyncio.TaskGroup() as tg:
tg.create_task(some_coro())
<h1>下面这行会报错!tg 已关闭</h1><p>tg.create_task(another_coro()) # RuntimeError!
-
TaskGroup实例不可复用,每次都需要新的async with块 - 不要把
tg保存为实例变量跨方法传递,它的生命周期严格绑定于上下文管理器 - 调试时可在创建 task 前加
assert not tg._closed(仅用于开发,非公开 API)
真正麻烦的不是语法,是判断哪些任务该放进同一个 TaskGroup:它们得共享相同的失败语义——要么全活,要么共死。分组边界划错,后面 debug 时看到的 ExceptionGroup 就像一锅乱炖。

















