asyncio.taskgroup 是 python 3.11 引入的结构化并发工具，替代 asyncio.gather，核心优势是异常时自动取消其余任务并确保资源清理，需在 async with 中使用 tg.create_task() 提交任务，异常以 exceptiongroup 形式抛出。
asyncio.TaskGroup 是什么，为什么不能用 asyncio.gather
asyncio.TaskGroup 是 Python 3.11 引入的正式替代 asyncio.gather 的结构化并发工具，核心价值在于「异常传播」和「自动取消」：只要一个任务出错，其余正在运行的任务会立即被取消，避免资源泄漏或静默失败。而 asyncio.gather 默认会等所有任务完成（除非加 return_exceptions=True），且无法保证未完成任务被及时清理。
典型误用场景：你启动了 5 个网络请求，第 3 个抛出 TimeoutError，但第 4、5 个还在发包——TaskGroup 能刹住车，gather 不会。
基本用法：with asyncio.TaskGroup() as tg: tg.create_task(...)
必须在 async with 上下文中使用，不能手动实例化后调用方法。所有任务必须通过 tg.create_task() 提交，不能直接用 asyncio.create_task()。
-
tg.create_task()返回的是asyncio.Task实例，可 await 获取结果，也可忽略（fire-and-forget） - 如果某任务 raise 异常，整个
with块会立即退出，其他任务收到asyncio.CancelledError - 即使所有任务都成功，也要等最慢的那个结束才退出
with
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/4a2dd4fd7b6db6c7600ecb27cc83a7c4">https://www.php.cn/link/4a2dd4fd7b6db6c7600ecb27cc83a7c4</a>"))
task2 = tg.create_task(fetch("<a href="https://www.php.cn/link/b7be11f1c3a8aba967e0a54a2fb2bd21">https://www.php.cn/link/b7be11f1c3a8aba967e0a54a2fb2bd21</a>"))
result1 = await task1 # 可以 await 单个
result2 = await task2 # 也可以一起 await
print(result1, result2)</p><p>asyncio.run(main())
</p>
捕获多个异常：try/except 包裹 TaskGroup 块
TaskGroup 抛出的异常不是单个错误，而是 ExceptionGroup（Python 3.11 新类型），里面可能包含多个子异常。直接 except Exception 捕不到，必须显式处理。
- 用
except* ValueError:可匹配所有子异常中的ValueError（注意是except*，不是except） - 用
except ExceptionGroup as eg:捕获整体，再用eg.exceptions查看全部 - 若只关心是否有错，不关心具体类型，用
except*最安全
async def risky_task(n):
if n == 2:
raise ValueError("bad value")
if n == 3:
raise RuntimeError("oops")
return n * 2
<p>async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(risky_task(1))
tg.create_task(risky_task(2))
tg.create_task(risky_task(3))
except<em> ValueError as eg:
print("Got ValueErrors:", eg.exceptions)
except</em> RuntimeError as eg:
print("Got Runtimes:", eg.exceptions)
</p>
常见坑：嵌套 TaskGroup、忘记 await、与 loop.run_in_executor 混用
TaskGroup 不支持嵌套：在 async with tg1: 里再开 async with tg2: 会报 RuntimeError: Task group is already running。需要并发嵌套？用普通 asyncio.create_task() + 手动 cancel 逻辑。
- 提交任务后不 await，也不存引用 → 任务可能被 GC 中断，尤其在短生命周期协程中
- 混用
loop.run_in_executor时，记得 executor 任务本身不参与 TaskGroup 生命周期管理，需自行 handle 异常和超时 - 在非 async 函数里调用
asyncio.run()再启 TaskGroup 是安全的；但在已运行的 event loop 里（如 Jupyter、FastAPI route）直接用需确认 loop 状态
最易被忽略的一点：TaskGroup 的「自动取消」只作用于它自己 create_task 出来的协程，对同步阻塞、subprocess.Popen 或第三方异步库（如 httpx.AsyncClient）内部状态无感知——那些还得靠 timeout 参数或信号机制兜底。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
