asyncio.TaskGroup 是 Python 3.11 引入的结构化并发机制，替代 gather() 和手动 create_task()，自动处理异常传播、取消和防孤儿任务；需在 async with 块内用 tg.create_task() 启动任务，不支持动态添加或 return_exceptions=True，多任务失败时抛出 ExceptionGroup。
Python 3.11 的 asyncio.TaskGroup 是替代 asyncio.gather() 和手动 create_task() + await 的更安全、更清晰的并发任务管理方式——它自动处理异常传播和取消，避免“孤儿任务”问题。
为什么不能继续用 asyncio.gather()？
asyncio.gather() 在某个子任务出错时，默认会取消其余正在运行的任务，但错误处理不透明；更关键的是，它无法在部分任务失败后仍让其他任务完成（除非显式传 return_exceptions=True），且不支持动态添加任务。而 TaskGroup 提供结构化并发语义：任一任务抛出未捕获异常，整个组立即取消，且异常被统一抛出到 with 块外，符合直觉。
常见错误现象：RuntimeWarning: coroutine 'xxx' was never awaited —— 这往往是因为用了 create_task() 却忘了 await 或没加到 gather 中；TaskGroup 强制你在上下文内启动任务，从语法上杜绝这类漏 await 问题。
- 使用场景：需要并发请求多个 API、并行处理文件、批量数据库写入等
- 性能影响：与
gather()相当，无额外开销；取消逻辑更精确，避免无效等待 - 兼容性：仅 Python 3.11+，旧版本需回退到
asyncio.create_task()+asyncio.wait()组合
如何正确使用 asyncio.TaskGroup 启动并发任务
核心是把任务创建放在 async with asyncio.TaskGroup() as tg: 块内，调用 tg.create_task()，而不是 asyncio.create_task()。
立即学习“Python免费学习笔记（深入）”；
示例：
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
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>"))
print(task1.result(), task2.result()) # 所有任务成功才执行到这里</p><p>asyncio.run(main())
注意：tg.create_task() 返回的是 asyncio.Task 对象，可后续调用 .result() 或 .exception()；但必须在 with 块结束后访问，否则可能仍在运行或已被取消。
- 不能在
TaskGroup外部调用tg.create_task()，会报RuntimeError: TaskGroup is closed - 若某任务抛出异常（如 HTTP 错误），整个
with块会提前退出，其余任务被取消，异常原样抛出 - 不支持传入
name=参数（不像create_task()），调试时建议在协程内部打日志
如何捕获单个任务异常而不中断整个组？
TaskGroup 的设计原则是“全成功或全失败”，所以它不提供类似 gather(return_exceptions=True) 的开关。如果确实需要容忍个别失败，得在任务内部处理异常，并返回标记结果：
async def fetch_safe(url):
try:
return await fetch(url)
except Exception as e:
return f"failed {url}: {e}"
<p>async def main():
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(fetch_safe("<a href="https://www.php.cn/link/e75a2d435455f0626ccf0af67216e76f">https://www.php.cn/link/e75a2d435455f0626ccf0af67216e76f</a>"))
task2 = tg.create_task(fetch_safe("<a href="https://www.php.cn/link/88b3febc5798a734026c82c1012408f5">https://www.php.cn/link/88b3febc5798a734026c82c1012408f5</a>"))
print(task1.result(), task2.result())
这种写法把错误“降级”为返回值，适合监控类、非关键路径任务。但要注意：这会让 TaskGroup 失去结构化错误传播能力，需自行判断哪些失败可忽略。
- 不要在
fetch_safe中静默吞掉异常（比如只写except:），否则调试时难以定位源头 - 若需区分错误类型（如重试网络错误、跳过解析错误），建议用不同返回结构（如
Result类或Union[Success, Failure]）
常见陷阱：嵌套 TaskGroup 和取消传播
可以嵌套使用 TaskGroup，但取消信号会逐层向外传播。例如外层组取消时，内层组也会被取消，且内层任务的取消异常（asyncio.CancelledError）不会被吞掉——除非你显式捕获。
容易踩的坑：
- 在
TaskGroup内部再用asyncio.create_task()：这些任务脱离组管理，可能成为孤儿，且不参与统一取消 - 在
tg.create_task()中传入已 await 过的协程（如tg.create_task(some_coro())而不是some_coro），会报TypeError: an asyncio.Future, a coroutine or an awaitable is required - 试图在
with块中对任务调用.cancel()：虽然可行，但通常没必要——组本身已具备取消语义；手动 cancel 可能导致重复取消警告
真正复杂的地方在于异常链的还原：当多个任务同时失败，TaskGroup 抛出的异常是 ExceptionGroup，需要用 except* ValueError:（PEP 654）或 exceptiongroup.ExceptionGroup 的 API 来匹配，这点容易被忽略。
