Python asyncio.TaskGroup组织并发任务与异常取消

Python 异步代码里，如果几个任务属于同一个业务请求，最稳妥的组织方式是把它们放进 asyncio.TaskGroup。这样上下文退出时会等待组内任务；某个任务抛出非取消异常时，其他任务会收到取消信号，最后由 ExceptionGroup 把失败结果交给调用方处理。它解决的不是“同时跑几个协程”这么简单，而是把任务的归属、取消和收尾放在同一个边界里。

官方文档：https://docs.python.org/3/library/asyncio-task.html

记住三个判断：关联任务用同一个 TaskGroup；取消清理写在 finally；异常要在上下文外用 except* 分类接住。

把关联任务放进同一个生命周期

asyncio.create_task() 只负责把协程调度成任务，调用方还要自己保存引用并等待它们。TaskGroup 把这层管理收拢到 async with 中，适合“用户信息和订单摘要必须一起完成”这类有明确边界的工作。

import asyncio

async def fetch_user(user_id: int) -> dict:
    try:
        await asyncio.sleep(0.2)  # 模拟异步 I/O
        return {"id": user_id, "name": "Lin"}
    finally:
        print("user task cleaned")  # 取消或正常结束都执行清理

async def fetch_orders(user_id: int) -> list[dict]:
    try:
        await asyncio.sleep(0.4)  # 让两个任务保持并发
        return [{"id": 1001, "user_id": user_id}]
    finally:
        print("orders task cleaned")  # 释放连接、锁等资源应放这里

async def load_profile(user_id: int) -> tuple[dict, list[dict]]:
    async with asyncio.TaskGroup() as group:
        user_task = group.create_task(fetch_user(user_id))
        orders_task = group.create_task(fetch_orders(user_id))
    # 离开上下文时两个任务都已结束，结果可以安全读取
    return user_task.result(), orders_task.result()

asyncio.run(load_profile(7))

这里的关键不是把两个 create_task() 写在一起，而是让它们共享同一个任务组。任意一个任务出现真正的业务异常，TaskGroup 会取消仍在运行的同组任务，并等待它们完成清理；只有上下文退出后，调用方才继续往下走。

让 CancelledError 完成资源收尾

任务被取消时，asyncio.CancelledError 会在下一个可中断的等待点抛出。数据库连接、文件句柄、临时锁或取消后的状态标记，都应该放在 finally 中处理。若为了记录日志捕获了 CancelledError，清理完成后通常还要继续抛出，不能把取消信号吞掉。

async def guarded_request(client, url: str) -> bytes:
    response = None
    try:
        response = await client.get(url)
        return await response.read()
    except asyncio.CancelledError:
        print("request cancelled")  # 只记录取消原因，不把取消改成成功
        raise
    finally:
        if response is not None:
            await response.close()  # 确保取消路径也释放响应资源

一个常见误区是写出 except Exception 作为“大兜底”。CancelledError 直接继承自 BaseException，通常不会被它捕获；真正危险的是显式捕获后返回默认值，或者在清理代码里再次长时间等待，导致整个任务组迟迟无法退出。

用 ExceptionGroup 接住失败结果

TaskGroup 不会把多个子任务的失败压成一条普通异常。所有任务结束后，非取消异常会组合成 ExceptionGroup。调用方可以在 async with 外用 except* 按类型拆开；处理器执行后，未处理的异常仍会继续向上抛出。

async def run_batch() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(load_config())
            group.create_task(sync_records())
    except* ValueError as error_group:
        print(f"输入数据失败：{len(error_group.exceptions)}")  # 只处理业务输入错误
    except* OSError as error_group:
        print(f"外部资源失败：{len(error_group.exceptions)}")  # 网络或文件错误单独告警

asyncio.run(run_batch())

如果第一个任务很快失败，另一个任务可能只收到 CancelledError，因此未必会贡献第二个业务异常。不要把“异常数量”当成任务数量；更可靠的判断是看异常类型和每个任务的清理日志。

超时、外部取消与嵌套任务组的排查清单

遇到“任务没有停”“异常没有进入处理器”或“退出上下文卡住”时，可以按下面顺序排查：

嵌套 TaskGroup 时，内层会先处理自己的子任务异常，外层再收到对应的取消和异常结果。工程上建议让每层只管理自己创建的任务，不要把同一个 TaskGroup 跨越多个不清晰的业务边界传递。

相关问题

TaskGroup 和 gather() 怎么选？需要强生命周期、失败后联动取消和结构化异常处理时选 TaskGroup；只是收集一组独立结果、并且已有明确的异常策略时，gather() 仍可以使用。

能不能在 TaskGroup 里捕获 CancelledError 后返回空结果？除非你明确设计了取消抑制并处理取消状态，否则不建议这样做。大多数业务应完成必要清理后重新抛出，让任务组和上层超时机制知道任务确实被取消。