asyncio.TaskGroup是Python 3.11+动态管理异步任务最安全方式，自动等待子任务、统一异常处理、支持运行时增删任务，且必须用async with语法。
用 asyncio.TaskGroup 动态管理异步任务列表，是 Python 3.11+ 最干净、最安全的方式——它自动等待子任务完成，且能统一处理异常，不用手动 asyncio.create_task() + await asyncio.gather() 那套容易漏收尾的组合。
为什么不能直接用 asyncio.create_task() 堆列表
很多人写成这样：
tasks = []
for url in urls:
tasks.append(asyncio.create_task(fetch(url)))
await asyncio.gather(*tasks)问题在于：如果某个 fetch() 抛出异常，gather() 默认会取消其余任务；更麻烦的是，你没法在任务中途动态增删——比如根据响应结果决定是否启动后续任务。而 TaskGroup 允许你在 with 块内随时调用 tg.create_task()，任务生命周期完全由上下文管理。
- 没用
TaskGroup时，任务可能“泄露”：忘记 await 或异常中断后残留未完成的Task对象 -
create_task()返回的Task对象一旦脱离引用，可能被静默丢弃（尤其在循环中反复覆盖变量） -
gather()不支持“部分失败继续执行”，而TaskGroup可以配合except*精确捕获多异常
TaskGroup 必须在 async with 中使用
这是硬性要求，不是建议。直接实例化 asyncio.TaskGroup() 不会启动上下文管理逻辑，也不会自动 await 子任务。
立即学习“Python免费学习笔记（深入）”；
- ✅ 正确写法：
async with asyncio.TaskGroup() as tg: - ❌ 错误写法：
tg = asyncio.TaskGroup()或await tg.__aenter__()（不保证行为一致） - 离开
async with块时，TaskGroup会自动等待所有已创建任务结束；哪怕某任务抛异常，也会等其他任务完成再汇总抛出
示例：
async def batch_fetch(urls):
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(fetch(u)) for u in urls]
return [t.result() for t in tasks]动态添加任务：在 async with 块内随时调用 tg.create_task()
这才是“动态管理”的核心能力。你可以在一个任务的回调里、在条件分支中、甚至在另一个 await 后继续加新任务——只要还在 async with 块里。
- 每个
tg.create_task()返回的Task对象可存入列表或字典，用于后续检查状态（如task.done()、task.exception()） - 不支持从外部取消单个任务（
task.cancel()会破坏TaskGroup的统一生命周期），如需取消，请改用asyncio.wait_for()包裹单个协程 - 若需按优先级调度，得靠协程内部逻辑控制，
TaskGroup本身不提供优先队列机制
常见场景：根据 API 响应分页拉取数据
async def fetch_all_pages(start_url):
results = []
async with asyncio.TaskGroup() as tg:
queue = [start_url]
while queue:
url = queue.pop(0)
task = tg.create_task(fetch(url))
# 这里可以 await 单个结果并决定是否追加新任务
resp = await task
results.append(resp)
if resp.next_page:
queue.append(resp.next_page) # 下次循环继续加
return results异常处理必须用 except*（Python 3.11+）
TaskGroup 在退出时，如果多个子任务分别抛出异常，会打包成 ExceptionGroup 抛出。普通 except Exception: 捕不到。
- ✅ 正确捕获：
except* aiohttp.ClientError:或except* Exception: - ❌ 普通
except会跳过，导致异常未处理而程序崩溃 - 如果只关心是否有任一失败，可用
task.exception() is not None逐个检查，但失去并发异常聚合优势
示例：
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(fetch("https://a.com"))
tg.create_task(fetch("https://b.com"))
except* aiohttp.ClientConnectorError as eg:
print(f"连接失败 {len(eg.exceptions)} 次")
except* Exception as eg:
print(f"其他异常 {eg}")真正难的是设计任务边界：哪些逻辑该放进同一个 TaskGroup，哪些该拆成独立上下文。比如长周期监控任务和短平快请求混在一起，前者卡住会影响后者超时判断——这时候就得拆开，而不是硬塞进一个 group 里。
