<p>TaskGroup 专为任务生命周期管理、取消传播和资源清理设计,必须用 async with 启动,通过 tg.create_task() 创建子任务,异常需用 except* 捕获 ExceptionGroup,结果需显式调用 task.result() 获取。</p>

TaskGroup 不是“比 gather 更好用”,而是解决 gather 压根不管的那部分问题:任务生命周期、取消传播、资源清理。如果你只求快速拿到结果、不 care 后台有没有僵尸请求,gather 还更轻量;但只要涉及 HTTP 客户端、数据库连接、文件锁或重试逻辑,TaskGroup 是唯一能守住底线的选择。
必须用 async with,不能直接实例化
TaskGroup 不是一个普通类,asyncio.TaskGroup() 或 tg = asyncio.TaskGroup().__enter__() 都会立刻抛 RuntimeError: TaskGroup is not running。它只接受一种合法入口:
- 必须写成
async with asyncio.TaskGroup() as tg: - 所有子任务只能通过
tg.create_task()启动,不能用asyncio.create_task() - 离开
async with块后,tg自动关闭,再调用tg.create_task()会报RuntimeError: TaskGroup is closed - 不能把
tg存到全局变量、类属性或传给另一个协程——生命周期一断,就失效
except* 是唯一能捕获异常的方式
TaskGroup 在多个子任务失败时抛出 ExceptionGroup,不是单个异常,也不是列表。用传统 except Exception: 会完全漏掉它,导致异常静默、清理逻辑跳过、连接泄漏。
- 必须用
except* ValueError:捕获特定类型异常 - 多个任务同时失败(比如 DNS + TLS 错误),
ExceptionGroup.exceptions里原样保留全部原始异常,不用自己拼上下文 - 不能依赖
return_exceptions=True那套“全塞进列表再遍历判断”的老路——TaskGroup 不返回结果列表,也不支持该参数
结果获取方式完全不同:别再依赖顺序和返回值
gather 返回一个按启动顺序排列的 list,而 TaskGroup 不返回任何东西。你得显式保存每个 Task 对象,再调用 .result() 获取结果——而且顺序不保证,也不能靠索引对齐业务含义。
立即学习“Python免费学习笔记(深入)”;
- 写法必须是:
t1 = tg.create_task(coroutine1()),t2 = tg.create_task(coroutine2()),然后分别await t1或t1.result() - 某个任务失败时,
t1.result()会直接抛出那个异常(不是包装在列表里) - 如果想统一处理成功/失败,得在外层用
except*,而不是在结果列表里isinstance(x, Exception)判定
取消传播不是“立刻退出”,而是“等清理完再报错”
TaskGroup 的“自动取消”是协作式、可中断的。它发 CancelledError,但会等待子任务自己执行 async with 或 try/finally 中的清理逻辑——这恰恰是它最不可替代的地方,也是最容易被忽略的复杂点。
- 若子任务里没检查
asyncio.current_task().cancelled(),或死循环中没 await 可取消点,async with块会卡住,直到它自己退出或超时 - HTTP 请求用了
async with httpx.AsyncClient()?TaskGroup 能确保 client 正确走完__aexit__,连接释放 - 对比 gather:一个任务出错,其余照跑,可能还在发第 5 次重试、解码 10MB JSON、写磁盘——这些都在白费资源


















