当前位置：首页 > 文章列表 > 文章 > python教程 > Python asyncio TaskGroup 怎么让子任务失败自动取消同组任务

Python asyncio TaskGroup 怎么让子任务失败自动取消同组任务

来源：17golang原创 2026-09-07 00:59:01 0浏览收藏

并发抓取、批量校验或并行写入时，常见需求是“一个子任务已经失败，剩下的同组任务就别继续了”。asyncio.TaskGroup 正是为这个边界设计的：第一个子任务抛出非 CancelledError 异常后，TaskGroup 会取消其余任务，等待它们结束，再把未取消的异常组成 ExceptionGroup 抛出。要点是，子任务必须认真处理取消清理，外层必须用 except* 接住聚合异常。

要点速览

TaskGroup 需要 Python 3.11 及以上版本。
首个真正失败的任务触发同组取消，取消本身不会被当作业务失败重新汇报。
不要吞掉 CancelledError；在清理完成后继续抛出，让 TaskGroup 能正常收尾。

TaskGroup 的失败传播先看一个最小例子

下面让一个任务立即失败，另外两个任务等待较长时间。你不需要手动保存任务列表，也不需要逐个调用 cancel()：

import asyncio

async def worker(name, delay, should_fail=False):
    try:
        await asyncio.sleep(delay)
        if should_fail:
            raise ValueError(f"{name} 的输入无效")
        return f"{name} 完成"
    finally:
        # 无论成功还是收到取消，都在这里释放本任务持有的资源
        print(f"{name} cleanup")

async def main():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(worker("校验A", 0.2, True))
            group.create_task(worker("校验B", 5))
            group.create_task(worker("校验C", 5))
    except* ValueError as errors:
        # except* 处理 ExceptionGroup 中匹配到的异常子集
        for error in errors.exceptions:
            print("业务失败：", error)

asyncio.run(main())

运行时，校验A 先抛出 ValueError，校验B 和 校验C 在下一次可取消点收到取消。它们仍会执行 finally，但不会被当成三个业务错误重新打印。这里的 except* 是专门匹配异常组的语法。

为什么其他子任务会收到取消

TaskGroup 的行为可以拆成三层：任务节点负责执行，取消信号负责收拢同组工作，异常组负责向外表达真正的失败。下面的静态关系图只展示这些对象之间的职责边界，不代表运行时序。

Python asyncio TaskGroup 中失败任务、取消信号、同组任务和清理逻辑的静态关系 — 图1：查看失败域、取消域和清理域，理解 TaskGroup 如何把同组任务收拢到同一个生命周期。

第一个非取消异常发生后，TaskGroup 会取消仍在运行的兄弟任务；如果 async with 的主体还没有退出，包含这个上下文的外层任务也可能在等待点收到取消。TaskGroup 会等待全部任务结束，再向外抛出异常组。因此，外层代码不应在看到第一个错误后立即假设其他任务已经完成。

取消是协作式的。子任务只有在下一个 await 等可取消点才会感知它。把清理写在 finally 中最稳妥；如果确实要捕获 CancelledError 记录日志，应在清理结束后重新抛出：

async def consume(queue):
    connection = await open_connection()
    try:
        return await queue.get()
    except asyncio.CancelledError:
        # 只记录取消，不把取消伪装成正常结果
        print("consume cancelled")
        raise
    finally:
        # 关闭连接，避免失败收尾时泄漏资源
        await connection.close()

ExceptionGroup 应该怎么接

帮助读者理解异常组如何保留业务异常、排除取消信号并交给 except* 分类处理。 — 图2：看清任务域、异常域与处理域之间的关系，区分业务异常和连带取消。

TaskGroup 等待结束后，会把没有被取消的异常组合起来。可以按异常类型拆分处理，也可以先记录完整的异常组，再统一转换成上层业务错误。注意：传统的 except ValueError 不能替代 except* ValueError 来匹配一组并发异常。

现象	正确判断	处理方式
一个任务抛出 ValueError	它是触发取消的业务异常	用 except* ValueError 分类记录
兄弟任务收到 CancelledError	它是收尾信号，不等于业务失败	清理后继续 raise
多个任务几乎同时失败	可能聚合为多个叶子异常	保留 ExceptionGroup 上下文再处理

async def run_batch():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(validate("订单A"))
            group.create_task(validate("订单B"))
    except* (ValueError, TypeError) as errors:
        # 这里只处理业务校验类异常，未匹配的异常仍会继续传播
        for error in errors.exceptions:
            print(type(error).__name__, error)

如果需要让调用方知道“批处理失败”，可以在 except* 中记录每个叶子异常后抛出新的业务异常；不要为了让流程变绿而直接 pass。异常组的价值就在于保留并发任务的失败集合。