Python asyncio.TaskGroup怎样让一组协程一起失败并统一收集异常为什么结果为空或不符合预期

我第一次把 asyncio.TaskGroup 放进批量任务里时，最先遇到的不是并发问题，而是“明明创建了几个任务，最后的结果字典却像是空的”。原因通常不是 TaskGroup 没有等待，而是某个任务抛出异常后，异常在离开 async with 时以 ExceptionGroup 继续向外抛出，后面的结果读取代码根本没有机会执行。

TaskGroup 适合“同一组任务要共同成功，否则整体进入失败处理”的场景。要统一收集异常，必须在 TaskGroup 外层用 except* 接住异常组；要保留未失败任务的值，则要提前保存任务对象，再按状态读取结果。

本文用一个小型“批量读取配置分片”示例说明这条边界。示例代码和输出是解释用的静态内容，不代表在本机已经运行。

为什么 TaskGroup 的结果会看起来为空

async with asyncio.TaskGroup() as tg 退出时会等待组内任务。组内第一次出现非 CancelledError 异常后，TaskGroup 会取消还没有结束的同组任务，等取消处理完成，再把失败集中成 ExceptionGroup 或 BaseExceptionGroup 抛出。

因此下面这种写法的问题不是“没有并发”，而是异常没有被接住：

async with asyncio.TaskGroup() as tg:
    task = tg.create_task(load_config())

# 如果上面的任务失败，这一行不会执行
print(task.result())  # 这里看不到结果

另一个容易误判的地方是取消。没有失败的任务不一定能正常返回值，它可能在下一次 await 处收到 CancelledError。所以“结果为空”至少要拆成三种情况：异常让上下文提前抛出、任务被取消、调用方没有保存任务对象。

先搭一个能观察结果的小项目

为了同时看到异常和成功结果，先保存每个任务对象。这个小项目把三个配置分片交给三个协程：一个返回成功值，两个分别抛出不同异常。真实项目中，worker 可以替换成数据库查询、HTTP 请求或文件读取。

import asyncio


async def worker(name: str, failure: str | None = None) -> str:
    # 让任务先让出事件循环，模拟一次异步 I/O 等待。
    await asyncio.sleep(0.05)
    if failure == "value":
        # ValueError 表示当前分片内容不符合业务格式。
        raise ValueError(f"{name} 的配置格式不正确")
    if failure == "runtime":
        # RuntimeError 表示读取依赖或运行环境出现故障。
        raise RuntimeError(f"{name} 的读取依赖不可用")
    return f"{name}:ok"


async def collect_configs() -> tuple[dict[str, str], list[str]]:
    tasks: dict[str, asyncio.Task[str]] = {}
    results: dict[str, str] = {}
    errors: list[str] = []

    try:
        async with asyncio.TaskGroup() as group:
            # 保存 Task 引用，退出上下文后才能区分成功、失败和取消。
            tasks["base"] = group.create_task(worker("base"))
            tasks["feature"] = group.create_task(worker("feature", "value"))
            tasks["region"] = group.create_task(worker("region", "runtime"))
    except* ValueError as error_group:
        # except* 只处理异常组中匹配的 ValueError 分支。
        errors.extend(f"ValueError: {item}" for item in error_group.exceptions)
    except* RuntimeError as error_group:
        # 不同异常类型可以在同一个异常组中分别归档。
        errors.extend(f"RuntimeError: {item}" for item in error_group.exceptions)

    for name, task in tasks.items():
        # 取消任务没有结果；失败任务的 exception() 只用于判断，不再次抛出。
        if not task.cancelled() and task.exception() is None:
            results[name] = task.result()

    return results, errors


async def main() -> None:
    results, errors = await collect_configs()
    print({"results": results, "errors": errors})


asyncio.run(main())

这里的关键不是把异常“吞掉”，而是把它们放到明确的收集结构中。except* 处理结束后，函数仍然可以遍历任务对象；成功任务可以读取 result()，失败和取消任务则被排除。

用 except* 统一收集异常和成功结果

这个例子中，两个失败任务是否都出现在异常组里，取决于它们在取消传播前是否已经抛出；TaskGroup 不承诺“所有任务都会执行到最后”。更稳妥的理解是：它保证同组任务的生命周期被一起管理，并把已经发生的非取消异常集中报告。

示意输出可以是下面这样，异常顺序不应作为业务判断依据：

{'results': {'base': 'base:ok'},
 'errors': [
   'ValueError: feature 的配置格式不正确',
   'RuntimeError: region 的读取依赖不可用'
 ]}

如果只看到一个错误，不一定是收集逻辑失效，另一个任务可能在异常传播时已经被取消。需要保留部分成功结果时，务必把“是否允许部分成功”写进业务规则，而不是默认把取消当成空字符串或默认值。

什么时候应该改用 asyncio.gather

如果任务彼此独立，一个任务失败不应该取消其他任务，可以考虑 asyncio.gather。默认的 return_exceptions=False 会把第一个异常传播给等待方，但其他可等待对象不会因此自动取消；设置 return_exceptions=True 后，异常会和普通返回值一起出现在结果列表中。

还要注意不要在协程里无条件吞掉 CancelledError。TaskGroup 依赖取消机制完成收尾；如果业务确实要捕获取消，也应在清理完成后继续传播取消状态。

排查结果为空时先看这三处

1. 看 async with 外层是否捕获了 ExceptionGroup，没有捕获时，后续结果处理不会运行。
2. 看是否保存了 create_task() 返回的任务对象；只保存协程函数调用，无法在结束后读取任务状态。
3. 看业务是否真的允许部分成功；被取消的任务没有结果，不要把它和“返回空值”混为一谈。

把这三点分开后，TaskGroup 的行为就比较清楚了：它不是一个自动把所有错误变成列表的收集器，而是一个有结构化并发边界的任务组。需要整体失败就用它，需要逐项容错就选 gather，并明确处理每个结果槽位。

相关问题

TaskGroup 是从哪个 Python 版本开始提供的？

官方文档标注为 Python 3.11 新增。如果运行环境更旧，需要升级 Python，或者使用已有的任务编排方式。

为什么 except* 不能写成普通 except？

普通 except 适合接收一个异常对象；TaskGroup 可能抛出包含多个异常的 ExceptionGroup，except* 才能按异常类型拆分其中的分支。