Python asyncio.TaskGroup 取消异常怎么收敛：从 gather 迁移到结构化并发

把 Python 异步代码从 asyncio.gather() 换成 asyncio.TaskGroup，真正需要重新确认的是失败传播：一个子任务出错后，其他任务是否应该继续、取消信号在哪里被清理、多个异常最后以什么形式交给调用方。TaskGroup 适合把这些约定收拢到一个作用域里，但它不是给 gather 换个名字。

先看失败语义：gather 和 TaskGroup 不再是同一条路

gather 更像一个结果收集器。默认情况下，一个 awaitable 抛出异常时，异常会尽快交给等待方；其他 awaitable 是否继续运行，不能简单按“整个批次都失败”理解。线上批处理里，这种行为很容易留下仍在写文件、占用连接或继续重试的兄弟任务。

TaskGroup 的边界更明确：同组任务里首次出现非 CancelledError 异常后，剩余任务会被取消，退出 async with 时再把非取消异常组合成 ExceptionGroup。因此迁移时，异常处理位置通常要从单个 except 调整为上下文管理器外层的 except*。

最小迁移写法：把任务创建放进同一个作用域

下面的例子模拟并发读取三个分片。旧代码可以用 gather 收集结果；迁移后用 TaskGroup.create_task 创建任务，并在上下文退出后读取每个 Task 的结果。

import asyncio

async def fetch_part(name: str, delay: float, failed: bool = False) -> str:
    try:
        await asyncio.sleep(delay)
        if failed:
            raise RuntimeError(f"{name} returned 503")
        return f"{name}:ok"
    finally:
        print(f"cleanup {name}")

async def load_all() -> list[str]:
    async with asyncio.TaskGroup() as group:
        tasks = [
            group.create_task(fetch_part("user", 0.05)),
            group.create_task(fetch_part("orders", 0.10, failed=True)),
            group.create_task(fetch_part("inventory", 0.30)),
        ]
    return [task.result() for task in tasks]

try:
    asyncio.run(load_all())
except* RuntimeError as errors:
    for error in errors.exceptions:
        print("upstream failure:", error)

这个写法有两个容易漏掉的细节。第一，tasks 列表只保存 Task 句柄，真正的异常会在离开 TaskGroup 时抛出，所以结果列表不会在失败路径上被错误地部分返回。第二，inventory 被取消时仍会进入 finally，清理连接、临时目录或指标计时器的代码应放在那里。

旧代码的三个迁移风险

把 CancelledError 当普通业务异常吞掉

CancelledError 直接继承自 BaseException。如果协程为了记录日志捕获它，应该在清理结束后重新抛出；否则 TaskGroup 等待退出时可能一直等不到真实的取消状态，超时和资源回收都会变得难查。

async def fetch_part(name: str, delay: float) -> str:
    resource = await open_resource(name)
    try:
        return await resource.read(delay)
    except asyncio.CancelledError:
        await resource.abort()
        raise
    finally:
        await resource.close()

仍然只捕获 Exception，而没有处理 ExceptionGroup

TaskGroup 退出时可能产生 ExceptionGroup。如果调用方需要把上游 503 转成可重试错误，可以使用 except* 精确处理 RuntimeError；不要为了“让代码先跑起来”直接把所有异常打平成字符串。

把 gather 的 return_exceptions=True 原样搬过来

return_exceptions=True 的调用方往往依赖“结果数组里混着异常对象”。TaskGroup 的默认目标是失败即取消同组任务，迁移后应重新定义返回协议：是让批次失败，还是在每个子任务内部把可预期业务失败转换成一个明确的结果对象。

回归检查：故障、取消和成功路径都要跑

不要只测所有任务成功的情况。最小回归集可以固定三个输入：第一个任务失败、最后一个任务失败、外部取消父任务。每个场景都记录任务结束状态，并确认被取消的任务执行了清理分支。

async def test_failure_cancels_siblings():
    try:
        await load_all()
    except* RuntimeError as errors:
        if len(errors.exceptions) != 1:
            raise RuntimeError("unexpected error count")

async def test_cleanup_log(log):
    # 断言 orders、inventory 的结束记录都已出现
    if not {"cleanup user", "cleanup orders", "cleanup inventory"} 

如果测试依赖日志顺序，要先区分“任务完成顺序”和“清理完成顺序”。取消发生后，短任务可能先写出清理日志；不要把一次偶然的输出排列当成 API 保证。

版本边界与迁移清单

TaskGroup 自 Python 3.11 加入。项目仍支持 Python 3.10 时，不能只在一个模块里替换调用，还要处理运行环境、依赖声明和测试矩阵。更稳的切法是先把任务函数的取消清理写正确，再按运行版本选择实现；不要让兼容层改变业务异常的含义。

• 确认运行时和 CI 至少覆盖项目声明的最低 Python 版本。
• 把 CancelledError 清理和重新抛出写进任务函数的测试。
• 为同组任务失败时的 ExceptionGroup 建立明确的转换边界。
• 删除对“异常对象混在结果数组中”的隐式依赖，改成可读的结果类型。
• 检查超时、父任务取消和资源关闭是否仍能在 finally 中完成。

相关问题

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

不会。它只会取消尚未完成的同组任务；已经完成的任务保留结果，退出时可以读取对应 Task 的 result()。

为什么 except* 不能和普通 except 混用处理同一个异常组？

except* 会按异常组中的类型拆分处理，语义和普通异常链不同。通常把 TaskGroup 的边界放在一个清晰的函数外层，在那里按业务异常类型分支，避免在深层协程里拆组。

什么时候继续使用 gather？

如果任务彼此独立，失败后仍希望其他任务继续，且调用方确实需要按输入顺序收集成功值和异常值，gather 仍然合适。选择标准是失败语义，而不是 API 新旧。

迁移完成的标志不是代码里出现了 TaskGroup，而是一次子任务失败后，剩余工作、取消清理和异常转换都符合团队约定。先把这三件事写进测试，再扩大替换范围，排查会比一次性重写整个并发层更可控。