Python异步任务管理：asyncio.TaskGroup全解析

Python 3.11 引入的 `asyncio.TaskGroup` 是动态管理异步任务的革命性方案——它通过 `async with` 语法自动统一等待所有子任务、精准聚合多异常（需配合 `except*`）、支持运行时灵活增删任务（如根据API响应动态分页拉取），彻底告别 `create_task() + gather()` 易漏收尾、无法中途干预、异常一崩全垮的痛点；它强制上下文管理，杜绝任务泄露，让高并发逻辑更健壮、可读、可维护，是现代Python异步编程不可或缺的核心工具。

用 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 子任务。

• ✅ 正确写法：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 里。

以上就是本文的全部内容了，是否有顺利帮助你解决问题？若是能给你带来学习上的帮助，请大家多多支持golang学习网！更多关于文章的相关知识，也可关注golang学习网公众号。