TaskGroup是Python 3.11引入的结构化并发机制，确保子任务共生死、异常原样抛出并聚合为ExceptionGroup；而asyncio.create_task仅调度协程，不绑定生命周期，易导致僵尸任务、资源泄漏及异常丢失。
TaskGroup 是什么，为什么不能用 asyncio.create_task 直接替代
TaskGroup 是 Python 3.11 引入的正式异步并发控制机制，核心作用是**结构化并发（structured concurrency）**：确保所有子任务在 TaskGroup 退出前完成或被取消，避免“幽灵任务”泄露。它不是 asyncio.create_task 的语法糖——后者只是把协程调度进事件循环，不提供生命周期绑定；而 TaskGroup 会自动等待所有子任务结束，并在任一子任务抛出未处理异常时，立即取消其余任务并聚合异常。
常见错误现象：asyncio.create_task 启动后忘记 await task 或没做 asyncio.gather 等待，导致任务实际没执行完、主协程就退出了；或者某个子任务崩溃，其他还在跑，逻辑错乱。
- 必须配合
async with使用，不能手动实例化后调用方法 - 不支持嵌套同级
TaskGroup（但可以嵌套不同层级的async with） - 子任务中若使用
asyncio.sleep(0)或await asyncio_yield()主动让出控制权，不会影响TaskGroup的等待逻辑
基本用法：启动多个并发任务并统一等待
最典型场景是并发请求多个 API、并行处理一批数据。直接用 async with asyncio.TaskGroup() as tg: 开启上下文，然后在内部调用 tg.create_task() 提交协程：
import asyncio
<p>async def fetch(url):
await asyncio.sleep(0.1) # 模拟网络延迟
return f"done: {url}"</p><p>async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(fetch("<a href="https://www.php.cn/link/e75a2d435455f0626ccf0af67216e76f">https://www.php.cn/link/e75a2d435455f0626ccf0af67216e76f</a>"))
task2 = tg.create_task(fetch("<a href="https://www.php.cn/link/88b3febc5798a734026c82c1012408f5">https://www.php.cn/link/88b3febc5798a734026c82c1012408f5</a>"))
task3 = tg.create_task(fetch("<a href="https://www.php.cn/link/6fe0fa22eca4dea0f85b883b75c76a34">https://www.php.cn/link/6fe0fa22eca4dea0f85b883b75c76a34</a>"))</p><h1>此处所有 task 已完成或被取消</h1><pre class='brush:python;toolbar:false;'>print(task1.result(), task2.result(), task3.result())asyncio.run(main())
立即学习“Python免费学习笔记（深入）”；
注意：tg.create_task() 返回的是 asyncio.Task 实例，可后续调用 .result() 或 .exception()；但不能在 async with 块外访问这些 task，因为它们可能已被清理。
- 所有子任务共享同一事件循环和上下文变量（
contextvars），这点和asyncio.create_task一致 - 如果某子任务抛出异常（非
BaseException），TaskGroup会立即取消其余任务，并将异常包装为ExceptionGroup - 不支持传入
name=参数（Python 3.12+ 才支持）
处理异常：ExceptionGroup 不是普通异常
当多个子任务分别抛出不同异常时，TaskGroup 不会只抛第一个，而是聚合为 ExceptionGroup。直接 except Exception: 捕获不到它——必须显式处理：
async def bad_task(n):
if n == 2:
raise ValueError("oops")
await asyncio.sleep(0.05)
return n * 2
<p>async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(bad_task(1))
tg.create_task(bad_task(2))
tg.create_task(bad_task(3))
except<em> ValueError as eg: # 注意是 except</em>，不是 except
print("Caught ValueError group:", eg.exceptions)
except* Exception as eg:
print("Other errors:", eg.exceptions)</p>关键点：
- 必须用
except*语法（PEP 654），普通except无法匹配ExceptionGroup -
eg.exceptions是一个元组，每个元素是原始异常实例，可单独检查类型或打印 - 若只想忽略某类异常（比如超时），可在子任务里
try/except吞掉，避免上抛到TaskGroup
和 asyncio.gather 的关键区别在哪
asyncio.gather 也能并发执行多个协程，但它不具备结构化语义：即使你 await gather(...)，如果中途某个协程被取消，其他协程仍可能继续运行（取决于是否响应取消信号）；而 TaskGroup 在任意子任务异常或取消时，会主动向所有子任务发送取消请求，并等待它们响应。
-
gather(return_exceptions=True)会把异常包进结果列表，但不会取消其他任务；TaskGroup默认行为是“失败即终止全部” -
gather支持timeout=参数；TaskGroup不支持超时控制（需自行用asyncio.wait_for包裹子协程） - 性能差异极小，两者底层都基于
create_task，但TaskGroup多一层异常聚合开销 - 兼容性：3.11+ 才有
TaskGroup；gather从 3.7 就稳定存在
真正容易被忽略的点是：如果你在子任务里用了 asyncio.to_thread 或 loop.run_in_executor，这些线程/进程任务**不会被 TaskGroup 自动取消**——因为取消信号只发给 asyncio Task，不穿透到 OS 线程。这种场景下，得靠超时或手动中断逻辑兜底。
