TaskGroup 是 asyncio 的结构化并发原语,仅保障子任务全部完成(含异常)后退出作用域,不提供取消策略、超时或重试;必须用 except* 捕获 ExceptionGroup,且子任务须通过 tg.create_task() 创建并严格绑定于 async with 块内。

TaskGroup 是 asyncio 的结构化并发原语,不是协程池或任务调度器
它只负责确保子任务全部完成(无论成功或异常)后才退出作用域,不提供取消策略、超时控制或重试逻辑。误把它当 asyncio.create_task 的语法糖来用,容易漏掉异常传播或提前退出。
常见错误现象:TaskGroup 内某个协程抛出未捕获异常,整个组立即崩溃,但调用方如果没用 except* 捕获,会看到 ExceptionGroup 被原样抛出,而非单个异常 —— 这和传统 try/except 行为不一致。
- 必须用
except* Exception(注意星号)捕获,不能用普通except Exception - 若只想处理某类异常(如
TimeoutError),需显式写except* TimeoutError - 多个子任务同时失败时,
ExceptionGroup会聚合所有异常,不会静默吞掉
用 async with TaskGroup() 启动协程,别手动 create_task
TaskGroup 的设计前提是“所有子任务生命周期严格绑定于其作用域”,一旦离开 async with 块,未完成的任务会被自动 cancel 并等待结束。手动调用 asyncio.create_task 然后丢进 TaskGroup 不仅无效,还会导致任务游离在组外,失去结构化保障。
正确做法是直接 tg.create_task(...):
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(fetch_url("https://a.com"))
task2 = tg.create_task(fetch_url("https://b.com"))
# ✅ 两个任务受 tg 管理:任一失败则 tg 退出,两者都完成才继续
- 不要写
asyncio.create_task(...)再赋值给变量,那任务已脱离TaskGroup控制 -
tg.create_task返回的是asyncio.Task实例,可 await 它获取结果,但 await 必须在async with块内 - 若需提前取消某个子任务,应调用
task.cancel(),而不是试图从外部 cancelTaskGroup
TaskGroup 不支持超时,需配合 asyncio.wait_for 或信号机制
TaskGroup 本身没有 timeout 参数,也不能在 async with 后加 timeout=...。想实现“最多等 5 秒,超时就放弃所有子任务”,得在外层套一层 asyncio.wait_for。
librosa 是一个用于音频和音乐分析的 Python 库。它提供了特征提取、频谱分析、节拍跟踪、起始检测和音频可视化的工具,基于 NumPy 和 SciPy 构建,用于科学音频计算。
错误写法:async with asyncio.TaskGroup(timeout=5) as tg: → 会报 TypeError: TaskGroup() got an unexpected keyword argument 'timeout'
- 正确方式:用
await asyncio.wait_for(asyncio.TaskGroup().__aenter__(), timeout=5)不可行(因__aenter__不可 await),应改用asyncio.wait_for包裹整个async with块 - 示例:
await asyncio.wait_for(your_task_group_coro(), timeout=5),其中your_task_group_coro是个封装了async with TaskGroup()的协程函数 - 注意:超时触发时,
TaskGroup会 cancel 所有活跃子任务,但 cancel 本身可能需要时间,实际退出可能略晚于 timeout 值
嵌套 TaskGroup 时,内层异常不会自动透传到外层
如果一个 TaskGroup 里又开了另一个 TaskGroup,内层抛出的异常会被内层组捕获并包装成 ExceptionGroup,除非显式 re-raise,否则外层看不到原始异常类型。
这容易导致错误分类失效 —— 比如你专门捕获 ConnectionError 做降级,但被嵌套组包了一层,变成 ExceptionGroup 就匹配不上。
- 若需透传,内层应使用
except* ConnectionError as eg:+raise eg.exceptions[0] from None(取第一个原始异常) - 更稳妥的做法是避免深度嵌套,把不同责任域的任务拆到同级
TaskGroup中管理 - 调试时可用
print(repr(eg))查看ExceptionGroup内部结构,别依赖str(eg)
真正难的不是写对语法,而是理解 TaskGroup 的边界在哪:它管生命周期,不管重试、熔断、优先级或资源配额。这些得靠上层逻辑补足,而不是指望它变全能。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










