Python ExceptionGroup与except*处理多个并发异常

在编写涉及并发任务（如 asyncio）或批量处理的程序时，一个经典的痛点是：当多个任务同时失败时，程序只能捕获到第一个遇到的异常，后续的错误往往会被掩盖或丢失，导致调试困难。Python 3.11 引入了 ExceptionGroup 和 except* 语法，彻底解决了这个问题。这套机制允许我们将多个异常打包在一起处理，或者精准地捕获其中的特定类型。

1. 理解异常组的本质

传统的 try-except 语句一次只能处理一个异常。一旦异常抛出，代码块立即终止。ExceptionGroup 本质上是一个异常容器，它将多个子异常对象存放在一个列表中，作为一个整体抛出。

定义一个简单的 ExceptionGroup：

# 创建一个包含两个不同错误的异常组
eg = ExceptionGroup(
    "批量处理发生错误",
    [
        ValueError("数值无效"),
        TypeError("类型错误")
    ]
)

# 抛出这个异常组
# raise eg

当上述代码执行时，Python 会显示一个特殊的回溯信息，明确指出这是一个 ExceptionGroup，并列出了其中包含的所有子异常。这意味着我们不会错过任何一个错误细节。

2. 使用 except* 语法拆解异常组

既然异常是一组抛出的，传统的 except ValueError 就无法直接匹配到被包裹在组里的 ValueError。Python 3.11 提供了 except*（读作 except star）语法，专门用于“部分捕获”异常组中的特定成员。

except* 的工作逻辑是：如果 try 块中抛出了一个异常组，Python 会遍历组内的所有异常。只要某个子异常匹配了 except* 后的类型，该子异常就会被从组中取出并处理，而剩下的不匹配的异常会继续抛出。

编写以下代码来演示 except* 的行为：

try:
    # 抛出一个包含 ValueError, TypeError, NameError 的组
    raise ExceptionGroup(
        "混合错误组",
        [
            ValueError("这是 ValueError"),
            TypeError("这是 TypeError"),
            NameError("这是 NameError")
        ]
    )
except* ValueError:
    print("捕获到 ValueError，已处理")
except* TypeError:
    print("捕获到 TypeError，已处理")

运行这段代码，你会发现程序并没有正常结束。这是因为 NameError 还在异常组里没有被处理。Python 会将剩余的未处理异常重新打包成一个 ExceptionGroup 继续向上抛出，导致程序崩溃。如果要处理所有异常，必须确保所有子异常都有对应的 except* 块，或者最后使用一个裸露的 except* Exception: 来兜底。

3. 异常分发流程可视化

理解 except* 如何“拆分”异常组非常关键。下面的流程图展示了当一个包含多种类型异常的组遇到多个 except* 块时的处理逻辑。

从图中可以看出，except* 并不是“拦截”整个流程，而是“过滤”特定的异常。只有当所有异常都被过滤处理完毕，程序才算完全恢复正常。

4. 获取捕获到的具体异常

在实际业务中，仅仅知道发生了 ValueError 是不够的，我们通常需要获取具体的错误信息。在 except* 块中，as 关键字赋值的变量不再是一个单一的异常对象，而是一个 ExceptionGroup 实例（其中只包含被匹配到的那些异常）。

查看以下代码，了解如何提取具体的错误信息：

try:
    raise ExceptionGroup(
        "数据校验失败",
        [
            ValueError("ID 不能为空"),
            ValueError("年龄必须大于 0"),
            TypeError("姓名必须是字符串")
        ]
    )
except* ValueError as e_group:
    # e_group 是一个只包含 ValueError 的新 ExceptionGroup
    print(f"捕获到 {len(e_group.exceptions)} 个数值错误")

    # 遍历取出每一个具体的错误对象
    for error in e_group.exceptions:
        print(f"具体错误: {error}")

输出结果将显示我们成功捕获了 2 个错误，并且可以分别打印出“ID 不能为空”和“年龄必须大于 0”。

5. 实战场景：并发任务中的异常处理

ExceptionGroup 最主要的应用场景是原生协程（asyncio）。在 Python 3.11+ 中，asyncio.TaskGroup（结构化并发）会在子任务失败时自动抛出 ExceptionGroup，而不是像旧的 gather 那样只抛出第一个异常。

构建一个模拟并发下载的示例：

import asyncio

async def download_file(url, should_fail=False):
    print(f"开始下载: {url}")
    await asyncio.sleep(0.1)
    if should_fail:
        raise ConnectionError(f"连接失败: {url}")
    print(f"下载完成: {url}")
    return f"{url} 的数据"

async def main():
    # 使用 TaskGroup 管理并发任务
    async with asyncio.TaskGroup() as tg:
        # 创建三个任务，其中两个会失败
        tg.create_task(download_file("url-A", False))
        tg.create_task(download_file("url-B", True))
        tg.create_task(download_file("url-C", True))
        # 注意：TaskGroup 会在所有任务结束后或第一个错误时抛出 ExceptionGroup

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except* ConnectionError as e_group:
        print("处理并发下载中的连接错误：")
        for err in e_group.exceptions:
            print(f"  - {err}")
    except Exception as e:
        print(f"捕获到其他未预期异常: {e}")

执行上述脚本，你会看到即使 url-A 下载成功，程序依然会报错退出。这是因为 TaskGroup 的默认策略是“全部成功才算成功”，或者“全部失败”。通过 except* ConnectionError，我们可以一次性收集并记录所有失败的 URL（url-B 和 url-C），而不是只捕获到第一个就停止。

6. 关键差异总结

为了快速区分传统异常处理与新的异常组处理，请参考下表。