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

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

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

第一个非取消异常发生后，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 应该怎么接

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

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。异常组的价值就在于保留并发任务的失败集合。

三个容易让取消语义失效的细节

1. 捕获后不重新抛出。 子任务吞掉 CancelledError 可能让 TaskGroup 一直等待，或者把任务误判成成功。
2. 把 TaskGroup 当作重试器。 它只负责结构化并发和失败收拢，不会自动重试失败任务；重试要放在明确的业务边界外。
3. 忽略 Python 版本。 TaskGroup 在 Python 3.11 加入，旧运行时不能直接使用这套 API；部署环境应先确认解释器版本。

排查时按这个顺序看：先确认触发异常的子任务，再确认兄弟任务是否进入 finally，最后检查外层是否用 except* 保留了异常组。这样能把“任务被取消”和“任务真的失败”分开。

相关问题

TaskGroup 会取消已经完成的任务吗？

不会。它只取消仍在运行的同组任务，已经完成的结果不会因为兄弟任务失败而回滚。

可以在子任务里捕获所有异常吗？

可以做局部处理，但不要用宽泛捕获把 CancelledError 一起吞掉；取消清理完成后应继续抛出。

TaskGroup 和 asyncio.gather 有什么区别？

TaskGroup 默认提供同组失败联动取消和异常聚合，更适合有共同生命周期的任务；gather 是否取消其他任务取决于调用方式和取消对象。