Python如何动态管理异步任务列表？详解asyncio.TaskGroup的使用方法

在Python异步编程中，把一堆协程丢到事件循环里跑只是第一步，真正麻烦的是如何管理这些任务：任务失败了怎么通知其他任务？动态加入新任务怎么处理？退出时如何保证资源不泄漏？在Python 3.11之前，开发者通常依赖asyncio.gather或者手动收集Task对象来处理这些问题，但两种方式都有明显短板。gather默认会吞掉部分异常，手动管理又容易漏掉取消和等待的逻辑。Python 3.11引入的asyncio.TaskGroup提供了结构化并发的解决方案，让一组任务的生死始终绑定在一起，本文就来详细讲解它的使用方法。

一、TaskGroup的基本用法与核心特性

TaskGroup是一个异步上下文管理器，通过async with语句使用。进入上下文后，你可以调用它的create_task方法创建任务；退出上下文时，TaskGroup会自动等待组内所有任务完成，任何一个任务抛出异常，其余任务都会被立即取消。这是它和gather最本质的区别：任务组是一个整体，要么一起成功，要么一起失败。

import asyncio

async def worker(name: str, delay: float):
    await asyncio.sleep(delay)
    print(f"任务 {name} 完成")
    return name

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker("A", 1))
        tg.create_task(worker("B", 2))
        tg.create_task(worker("C", 0.5))
    # 走到这里时，三个任务全部执行完毕
    print("所有任务已完成")

asyncio.run(main())

这段代码的执行顺序是：C先完成，然后是A，最后B完成，接着上下文退出，打印“所有任务已完成”。注意async with块内部本身就是异步的，你可以在创建任务之后继续执行其他await操作，不必等任务全部结束才离开代码块。上下文管理器的__aexit__会负责收尾，等待所有任务结束后才真正退出，这就是所谓的结构化并发：任务的生存范围严格限定在代码块之内，不会出现游离在外的孤儿任务。

另外一个细节是，create_task返回的就是普通的asyncio.Task对象，你可以保存引用并在后续读取结果，比如task = tg.create_task(worker("A", 1))，之后调用task.result()获取返回值。但要注意，如果任务已经失败并且异常在TaskGroup层面被抛出过，再读取结果会重新抛出异常。

二、动态添加任务与外部引用TaskGroup

实际项目中，任务往往不是一开始就确定的。比如爬虫场景下，初始任务产生新链接，需要持续往组里追加下载任务，直到队列清空。TaskGroup完全支持这种动态模式，关键做法是把TaskGroup对象保存到外部变量，在任意协程内部继续调用它的create_task方法。

import asyncio

async def fetch(queue: asyncio.Queue, tg: asyncio.TaskGroup):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"处理: {item}")
        # 模拟处理过程中动态发现新任务
        if item < 3:
            await queue.put(item + 1)
            tg.create_task(fetch(queue, tg))

async def main():
    queue = asyncio.Queue()
    await queue.put(1)
    async with asyncio.TaskGroup() as tg:
        # 启动初始消费者
        tg.create_task(fetch(queue, tg))
        # 放入结束信号
        await queue.put(None)
    print("队列处理完毕")

asyncio.run(main())

上面代码展示了动态管理的核心思路：初始只启动一个消费者任务，后续任务在运行过程中按需追加。需要特别注意的是，动态模式下退出条件要设计好，否则TaskGroup在async with退出时会一直等待所有任务结束，造成程序挂起。常见的错误是在循环外追加任务导致永远有新任务产生，形成死循环。

还有一种模式是使用一个协调者协程，由它负责决定何时创建新任务、何时停止。这样TaskGroup只在main函数里创建一次，其他模块通过参数传递拿到引用，代码结构更清晰。如果你的任务来源是消息队列或者WebSocket推送，这种“协调者+任务组”的组合非常实用。

三、异常处理机制：ExceptionGroup详解

TaskGroup的异常处理和传统方式差异很大。当组内多个任务同时失败时，TaskGroup不会只抛出第一个异常，而是抛出一个ExceptionGroup（准确说是asyncio.exceptions.ExceptionGroup的别名），把所有失败的异常打包在一起。这解决了长期困扰Python异步开发的问题：gather在return_exceptions为False时只报告第一个异常，其余异常信息直接丢失。

import asyncio

async def failing(name: str):
    await asyncio.sleep(0.5)
    raise ValueError(f"{name} 出错了")

async def slow():
    await asyncio.sleep(10)
    print("这行不会执行，因为会被取消")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(failing("任务1"))
            tg.create_task(failing("任务2"))
            tg.create_task(slow())
    except* ValueError as eg:
        # except* 语法专门用于捕获异常组
        for exc in eg.exceptions:
            print(f"捕获到: {exc}")

asyncio.run(main())

这里有两个关键点值得展开。第一，slow任务本来要睡10秒，但因为它同组的两个任务在0.5秒后失败，TaskGroup会取消所有未完成的任务，程序几乎立即退出，这就是“一损俱损”的取消联动。第二，捕获异常组要使用except*语法（Python 3.11新增），它会匹配异常组中指定类型的子异常，eg.exceptions是一个包含所有匹配异常的列表。如果用普通的except ValueError，将无法捕获到ExceptionGroup本身，因为异常组的类型不是ValueError。

需要区分的是，如果任务内部自己捕获了异常（比如worker函数里写了try/except），那这个失败不会传导到TaskGroup层面，其他任务照常运行。只有异常真正从任务中逃逸出来，才会触发整组取消。这个设计给了开发者灵活控制的能力：可恢复的错误在任务内部处理，致命错误才向上抛出让整组终止。

四、TaskGroup与gather、create_task的对比

选型时可以从几个维度对比这三种方式。asyncio.create_task最灵活但最危险，任务创建后就成了“放出去的风筝”，忘记保存引用可能被垃圾回收，忘记await可能悄悄丢失异常。gather适合批量等待固定数量的协程，写法简洁，但异常处理粗糙，且不会自动取消兄弟任务。而TaskGroup牺牲了一点灵活性（必须配合async with使用），换来了严格的生存期保证和完整的异常信息。

从表格可以看出，如果你的运行环境是Python 3.11及以上，绝大多数需要并发等待的场景都应该优先使用TaskGroup。唯一需要斟酌的场景是你明确希望某个任务失败后其他任务继续执行，这时可以在任务内部做好异常隔离，或者退回使用gather(return_exceptions=True)。

还有一个进阶技巧：TaskGroup支持嵌套。你可以在一个TaskGroup的上下文中再开一个子TaskGroup，形成任务层级树。子组的异常会被包装进父组的ExceptionGroup层层上传，取消信号也会从父组向子组传播。这种嵌套结构非常适合表达复杂的任务依赖关系，比如一个服务模块内部管理自己的并发任务，模块整体又作为一个任务挂在全局任务组上。

五、实战注意事项与常见坑

第一个常见的坑是在async with块外面调用create_task，会直接抛出RuntimeError。TaskGroup对象只在上下文活跃期内可用，退出后再创建任务是不允许的。如果你需要长期运行的后台任务管理，应该考虑把TaskGroup的生存期拉长到整个应用生命周期，或者改用更重量级的方案比如独立的管理器类。

第二个坑是取消时的清理逻辑。当TaskGroup取消任务时，每个任务内部会收到CancelledError，如果你的任务占用了网络连接、文件句柄等资源，务必用try/finally确保释放。协程内的finally块在取消时依然会执行，但要避免在finally中再执行耗时的await操作，否则可能触发二次取消。

import asyncio

async def resource_task():
    conn = None
    try:
        conn = await open_connection()
        await asyncio.sleep(100)
    finally:
        # 取消时也会执行清理
        if conn:
            await conn.close()

async def open_connection():
    await asyncio.sleep(0.1)
    return object()

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(resource_task())
            raise RuntimeError("外部主动失败")
    except* RuntimeError:
        print("已触发整组取消，资源已清理")

asyncio.run(main())

第三个注意点与Python版本相关。TaskGroup是3.11的新特性，如果你的项目还需要支持3.8、3.9、3.10，可以使用第三方的anyio库或者asyncio-taskgroup向后移植包，它们的API基本一致，迁移成本很低。等到项目升级到3.11后，把import换回标准库即可无缝过渡。

总结一下，asyncio.TaskGroup用上下文管理器的方式重新定义了Python异步任务的组织形式：生存期明确、失败联动、异常完整。无论是固定的一批任务还是动态增长的任务列表，它都比裸用create_task和gather更安全。如果你还在维护散落各处的Task引用列表，不妨趁早迁移到TaskGroup，代码的可读性和健壮性都会有明显提升。

Python异步编程asyncio.TaskGroup异步任务管理修改时间：2026-09-06 08:36:45