asyncio TaskGroup 失败怎么配置或排查

在 asyncio.TaskGroup 里，一个子任务失败后，程序通常不会像普通函数那样只抛出一个异常：同组未完成任务会先被取消，全部任务收尾后再把非取消异常聚合起来。排查“TaskGroup 失败”时，优先检查三件事：是否在 async with 内创建任务、是否用 except* 处理 ExceptionGroup、清理代码是否错误吞掉了 CancelledError。

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

稳定做法是让 TaskGroup 负责并发任务的生命周期，让业务层用 except* 分类错误，让每个协程在 finally 完成清理后继续传播取消。不要用空的 except Exception 把所有失败压成一条日志，也不要在取消分支里无限重试。

先确认失败发生在任务、任务组还是清理阶段

先用一个故意失败、一个可取消、一个正常收尾的最小组合复现。这样能看清异常是由任务本身抛出，还是任务组退出时聚合，或者是清理函数没有响应取消。

按 TaskGroup 的失败规则组织创建和等待

任务应在 async with asyncio.TaskGroup() 作用域内创建，离开作用域时由任务组统一等待。只要某个任务抛出非 CancelledError 异常，其他未完成任务会收到取消请求；这不是“随机少跑了几项”，而是 TaskGroup 的失败快速收敛规则。

import asyncio

async def fetch_one(name: str, delay: float, fail: bool = False) -> str:
    try:
        await asyncio.sleep(delay)
        if fail:
            raise ValueError(f"{name} 返回了不可用数据")
        return f"{name}: ok"
    finally:
        # 清理必须短且可取消，避免任务组退出被拖住
        await asyncio.sleep(0)

async def main() -> None:
    async with asyncio.TaskGroup() as group:
        # 在上下文内创建任务，退出时由 TaskGroup 统一等待
        group.create_task(fetch_one("cache", 0.05, fail=True), name="cache")
        group.create_task(fetch_one("profile", 0.20), name="profile")
        group.create_task(fetch_one("quota", 0.30), name="quota")

asyncio.run(main())

这段代码不会返回一个“失败任务列表”，而是让上下文退出时抛出异常组。不要在创建任务后立刻逐个 await task 来替代任务组管理；那会把生命周期、取消和异常聚合重新分散到调用方。

用 except* 拆分 ExceptionGroup 中的业务错误

普通 except ValueError 针对的是单个异常对象，而 TaskGroup 可能抛出 ExceptionGroup。Python 3.11 起可以用 except* 匹配异常组中的同类成员，分别记录可恢复错误和未知错误。

import asyncio

async def run_batch() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            # 两类失败交给上层按类型分类，而不是提前吞掉
            group.create_task(load_profile(), name="profile")
            group.create_task(load_quota(), name="quota")
    except* ValueError as errors:
        # 业务输入问题可记录后转人工或补偿，不要盲目重试
        for error in errors.exceptions:
            print(f"业务数据失败: {error}")
    except* TimeoutError as errors:
        # 只有明确是瞬时超时，才进入有限次数的重试队列
        print(f"瞬时失败数量: {len(errors.exceptions)}")

async def load_profile() -> None:
    await asyncio.sleep(0)
    raise ValueError("profile 字段缺失")

async def load_quota() -> None:
    await asyncio.sleep(0)
    raise TimeoutError("quota 服务响应超时")

except* 分支处理的是异常组中匹配的子集，未匹配部分仍会继续抛出。因此未知异常不要在最后随意静默；可以让它继续冒泡，由任务入口记录完整 traceback 和任务名。

清理时保留 CancelledError 的传播

TaskGroup、asyncio.timeout() 等结构化并发组件会使用取消完成内部控制。如果协程捕获 asyncio.CancelledError 后直接返回，外层可能误以为任务正常结束，或者在关闭阶段等待更久。正确做法是在 finally 释放资源；如果确实捕获取消，也要在清理完成后重新抛出。

import asyncio

async def worker(resource) -> None:
    try:
        await resource.run()
    except asyncio.CancelledError:
        # 记录取消上下文，但不要把取消伪装成成功
        resource.mark_cancelled()
        raise
    finally:
        # close 应该可重复调用，并且不要在这里启动无限重试
        await resource.close()

如果业务确实要把取消转换成自己的结果，必须非常明确地处理取消状态，并确认不会破坏上层超时、父任务取消或嵌套 TaskGroup。大多数网络请求、文件操作和队列消费场景，都应选择“清理后继续抛出”。

把失败处理接入日志、重试和通知门禁

生产代码可以把一次 TaskGroup 运行看成一个工作流门禁：任务名和输入标识进入日志；异常组按类型进入重试、补偿或告警；取消原因单独记录。重试只针对已确认的瞬时错误，并设置上限和退避，不要对 ValueError、权限错误或格式错误重复提交。

建议至少保留四个字段：任务名、异常类型、是否由兄弟任务取消、清理是否完成。这样看到“很多任务都 CancelledError”时，能回溯真正的首个失败，而不是把连带取消误报成几十个独立故障。任务组退出后再发送一次汇总通知，避免每个子任务单独报警造成噪声。

常见问题

TaskGroup 失败为什么不是普通异常？

因为组内可能有多个非取消异常，TaskGroup 会等任务收尾后把它们组合成 ExceptionGroup 或 BaseExceptionGroup。

能不能在每个任务里捕获所有 Exception？

可以记录上下文，但不要把取消也当作普通失败吞掉。对任务组出口仍应保留按类型分类的处理，让未知异常继续暴露。

为什么一个任务失败后其他任务也被取消？

这是 TaskGroup 的结构化并发语义：首个非取消异常触发同组剩余任务取消，任务组等待它们完成清理，再抛出聚合异常。

取消后必须重新抛出 CancelledError 吗？

通常是。除非你有清晰的取消转换设计，否则清理完成后重新抛出，才能让父任务、超时和嵌套任务组收到真实控制信号。