当前位置：首页 > 文章列表 > 文章 > python教程 > Python asyncio.TaskGroup 取消后为什么还有异常：ExceptionGroup 拆解与重试边界

Python asyncio.TaskGroup 取消后为什么还有异常：ExceptionGroup 拆解与重试边界

来源：17golang原创 2026-08-24 19:01:54 0浏览收藏

线上批量请求通常不是一个任务：读取用户配置、请求两个下游、写入审计记录，往往会被放进同一个 asyncio.TaskGroup。最容易误判的场景是其中一个子任务抛错后，其他任务看起来“只是被取消”，但外层捕获到的却不是那个原始异常，而是一组异常。这个行为不是 TaskGroup 多报了一次错，而是结构化并发在退出时保留了完整的失败边界。

TaskGroup 中一个任务失败后，兄弟任务会收到取消；离开上下文时，未被取消吞掉的失败会组成 ExceptionGroup。要重试，先用 except* 分出可重试异常，不能把整组异常一股脑重跑。

要点速览

第一个失败任务会触发同组兄弟任务取消，取消本身通常不是业务失败。
多个异常会以 ExceptionGroup 交给调用方，普通 except 不能精确拆分其中的类型。
except* 处理的是异常组中的匹配分支，处理后剩余分支仍会继续向外传播。
重试应只覆盖幂等且可恢复的分支，并给每一轮设置新的超时与尝试次数。

先复现一个“取消后还有异常”的现场

下面的实验让 worker-a 立即失败，另外两个任务分别等待和主动抛出。为了观察取消传播，取消分支只记录一次，然后继续抛出 CancelledError，不把取消当成成功。

import asyncio

async def worker_a():
    await asyncio.sleep(0.05)
    raise ValueError("invalid profile")

async def worker_b():
    try:
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("worker-b cancelled")
        raise

async def worker_c():
    await asyncio.sleep(0.08)
    raise TimeoutError("audit service timeout")

async def main():
    async with asyncio.TaskGroup() as group:
        group.create_task(worker_a())
        group.create_task(worker_b())
        group.create_task(worker_c())

asyncio.run(main())

运行时通常能先看到 worker-b cancelled，随后主协程退出并打印一个 ExceptionGroup。这里有两个观察点：TaskGroup 不会等一个已经失败的兄弟任务继续正常收尾；同时，取消传播与 ValueError、TimeoutError 这样的业务异常属于不同语义。

TaskGroup 的退出顺序到底发生了什么

可以把一次退出拆成三步。第一步，某个子任务出现非取消异常，TaskGroup 记录它，并开始取消尚未完成的兄弟任务。第二步，TaskGroup 等待这些任务结束；如果它们正确传播取消，通常不会凭空增加一条业务错误。第三步，所有任务完成后，TaskGroup 把需要向外报告的异常按层级组成 ExceptionGroup 或 BaseExceptionGroup。

因此，下面这种写法并不能稳定地得到单个异常：

try:
    async with asyncio.TaskGroup() as group:
        group.create_task(worker_a())
        group.create_task(worker_c())
except ValueError:
    print("retry")

如果外层拿到的是包含多个分支的 ExceptionGroup，普通 except ValueError 不负责从组中筛出 ValueError。更重要的是，直接重试整个 TaskGroup 还可能把已经完成的副作用重复提交。

用 except* 把可恢复分支拆出来

Python 3.11 引入的 except* 会匹配异常组中的子树。它适合把输入错误、下游超时和不可恢复错误分开处理：

async def run_batch():
    async with asyncio.TaskGroup() as group:
        group.create_task(worker_a())
        group.create_task(worker_c())

try:
    await run_batch()
except* TimeoutError as errors:
    for error in errors.exceptions:
        print("retryable:", error)
except* ValueError as errors:
    for error in errors.exceptions:
        print("reject input:", error)

每个 except* 分支处理自己的匹配结果。没有被处理的异常，会在处理完后重新组合并继续向外抛出。不要在 except* 分支里再写一个覆盖全部异常的“兜底重试”，否则输入校验失败也会被误当成临时网络问题。

TaskGroup 中 worker-a 失败后，兄弟任务收到取消，父协程得到 ExceptionGroup 的工程证据图

重试边界要放在任务之外

TaskGroup 负责一轮并发任务的生命周期，重试负责重新发起一轮业务操作。两者不要混成“某个子任务 catch 后自己循环”。更稳妥的做法是由外层按异常类型决定下一轮，同时限制次数、退避时间和幂等范围：

async def call_with_retry(max_attempts=3):
    for attempt in range(1, max_attempts + 1):
        try:
            await run_batch()
            return
        except* TimeoutError as errors:
            if attempt == max_attempts:
                raise
            print(f"timeout group, retry {attempt + 1}")
            await asyncio.sleep(0.2 * attempt)
        except* ValueError:
            raise RuntimeError("input is not retryable")

示例表达的是边界，不是万能模板。真实项目还要确认 run_batch() 里的写入是否幂等、超时任务是否可能已在服务端完成，以及每次重试是否需要重新生成请求标识。如果一个子任务已经产生不可逆副作用，就应把它从可自动重试集合里移出。