Python 3.11+的ExceptionGroup如何与asyncio.gather配合捕获

来源：建站技术作者：小黄人头衔：程序员

导读：本期聚焦于小伙伴创作的《Python 3.11+的ExceptionGroup如何与asyncio.gather配合捕获》，敬请观看详情，探索知识的价值。以下视频、文章将为您系统阐述其核心内容与价值。如果您觉得《Python 3.11+的ExceptionGroup如何与asyncio.gather配合捕获》有用，将其分享出去将是对创作者最好的鼓励。

在Python 3.11及更高版本中，ExceptionGroup作为新增的异常类型，专门用于处理多个异常同时出现的场景，而asyncio.gather是异步编程中并发执行多个任务并收集结果的常用工具，两者配合可以高效处理异步任务中多个异常同时抛出的情况。

ExceptionGroup基础特性

ExceptionGroup允许将多个异常打包成一个异常对象抛出，它的exceptions属性可以获取内部包含的所有异常实例。在Python 3.11之前，如果多个异步任务同时抛出异常，往往只能捕获到第一个抛出的异常，其余异常会被忽略，而ExceptionGroup解决了这个问题。

以下是一个简单的ExceptionGroup示例：

try:
    # 手动创建ExceptionGroup抛出多个异常
    raise ExceptionGroup(
        "多个异常发生",
        [ValueError("数值错误"), TypeError("类型错误"), RuntimeError("运行时错误")]
    )
except ExceptionGroup as e:
    print(f"异常组消息: {e.message}")
    print(f"包含的异常数量: {len(e.exceptions)}")
    for idx, exc in enumerate(e.exceptions, 1):
        print(f"第{idx}个异常: {type(exc).__name__}, 信息: {exc}")

asyncio.gather的默认异常行为

asyncio.gather用于并发运行多个异步任务，默认情况下，当其中一个任务抛出异常时，gather会立即取消其他尚未完成的任务，并且只会抛出第一个任务的异常，其余异常会被丢弃。这种行为和ExceptionGroup的特性并不匹配，需要手动调整参数才能配合。

先看默认行为的示例：

import asyncio

async def task1():
    await asyncio.sleep(0.1)
    raise ValueError("任务1抛出数值错误")

async def task2():
    await asyncio.sleep(0.2)
    raise TypeError("任务2抛出类型错误")

async def main():
    try:
        # 默认参数下，gather遇到第一个异常就取消其他任务，只抛出第一个异常
        await asyncio.gather(task1(), task2())
    except ValueError as e:
        print(f"捕获到异常: {e}")
    except TypeError as e:
        print(f"捕获到异常: {e}")

asyncio.run(main())

运行上述代码只会输出任务1的ValueError，任务2的TypeError不会被捕获，因为任务2在任务1抛出异常后被取消了。

配合ExceptionGroup捕获多异常的方法

要让asyncio.gather和ExceptionGroup配合，需要给gather传递return_exceptions=False之外的参数吗？不对，实际上需要设置return_exceptions=True吗？也不是，正确的方式是设置return_exceptions=False的同时，结合Python 3.11+的异常组机制？不对，正确的做法是为asyncio.gather设置return_exceptions=False时，当多个任务抛出异常，Python 3.11+的asyncio.gather会自动将多个异常打包成ExceptionGroup抛出，不过需要注意return_exceptions参数的取值。

关键参数说明

asyncio.gather的return_exceptions参数有两个取值：

False（默认）：任务抛出异常时，gather会取消其他任务，并且如果多个任务都抛出了异常，在Python 3.11+中会将这些异常打包成ExceptionGroup抛出，而不是只抛第一个。
True：不会取消其他任务，异常会作为结果返回，不会抛出ExceptionGroup，需要手动遍历结果处理异常。

正确配合的代码示例

以下示例展示如何设置参数让asyncio.gather抛出ExceptionGroup，然后捕获处理：

import asyncio

async def task1():
    await asyncio.sleep(0.1)
    raise ValueError("任务1数值错误")

async def task2():
    await asyncio.sleep(0.1)
    raise TypeError("任务2类型错误")

async def task3():
    await asyncio.sleep(0.2)
    return "任务3完成"

async def main():
    try:
        # 不设置return_exceptions，使用默认False，Python 3.11+下多个异常会组成ExceptionGroup
        results = await asyncio.gather(task1(), task2(), task3(), return_exceptions=False)
    except ExceptionGroup as e:
        print(f"捕获到异常组，消息: {e.message}")
        print("内部异常列表:")
        for exc in e.exceptions:
            print(f"  - {type(exc).__name__}: {exc}")
    else:
        print(f"所有任务完成，结果: {results}")

asyncio.run(main())

运行上述代码，会捕获到包含ValueError和TypeError的ExceptionGroup，任务3因为前两个任务同时抛出异常，在默认参数下会被取消，所以不会返回结果。

使用return_exceptions=True的替代方案

如果不想使用ExceptionGroup，也可以设置return_exceptions=True，此时所有任务都会执行完成，异常会作为结果返回，不会抛出异常，需要手动遍历结果处理：

import asyncio

async def task1():
    await asyncio.sleep(0.1)
    raise ValueError("任务1数值错误")

async def task2():
    await asyncio.sleep(0.1)
    raise TypeError("任务2类型错误")

async def task3():
    await asyncio.sleep(0.2)
    return "任务3完成"

async def main():
    # 设置return_exceptions=True，异常会作为结果返回，不会抛出
    results = await asyncio.gather(task1(), task2(), task3(), return_exceptions=True)
    for idx, res in enumerate(results, 1):
        if isinstance(res, Exception):
            print(f"任务{idx}发生异常: {type(res).__name__}, 信息: {res}")
        else:
            print(f"任务{idx}结果: {res}")

asyncio.run(main())

这种方式不会抛出ExceptionGroup，适合不需要异常组特性的场景，但是不符合我们配合ExceptionGroup的需求。

注意事项

ExceptionGroup是Python 3.11新增的特性，低于该版本的Python无法使用，使用时需要确认运行环境版本。
当asyncio.gather设置return_exceptions=False时，只有在多个任务同时抛出异常的情况下才会生成ExceptionGroup，如果只有一个任务抛出异常，还是会抛出单个异常，捕获时需要同时处理两种场景。
ExceptionGroup支持嵌套，如果内部异常本身也是ExceptionGroup，可以通过递归的方式遍历所有底层异常。

以下是兼容单个异常和ExceptionGroup的捕获示例：

import asyncio

async def task1():
    await asyncio.sleep(0.1)
    raise ValueError("任务1数值错误")

async def task2():
    await asyncio.sleep(0.1)
    raise TypeError("任务2类型错误")

async def single_task():
    await asyncio.sleep(0.1)
    raise RuntimeError("单个任务异常")

async def handle_exception(e):
    # 递归处理嵌套的ExceptionGroup
    if isinstance(e, ExceptionGroup):
        for sub_exc in e.exceptions:
            await handle_exception(sub_exc)
    else:
        print(f"处理异常: {type(e).__name__}, 信息: {e}")

async def main():
    # 测试多个任务异常的场景
    try:
        await asyncio.gather(task1(), task2(), return_exceptions=False)
    except Exception as e:
        await handle_exception(e)

    print("---分割线---")

    # 测试单个任务异常的场景
    try:
        await asyncio.gather(single_task(), return_exceptions=False)
    except Exception as e:
        await handle_exception(e)

asyncio.run(main())

总结

Python 3.11+的ExceptionGroup和asyncio.gather配合的核心是正确设置gather的return_exceptions参数，默认参数下多个任务同时抛出异常会自动生成ExceptionGroup，捕获后可以通过其exceptions属性获取所有异常。如果需要所有任务都执行完成再处理异常，可以设置return_exceptions=True手动遍历结果。开发者可以根据实际场景选择合适的处理方式，提升异步代码的异常处理能力。

Python ExceptionGroup asyncio_gather 异步编程修改时间：2026-06-29 23:15:42

免责声明：已尽一切努力确保本网站所含信息的准确性。网站内容多为原创整理与精心编撰，观点力求客观中立。本站旨在免费分享，内容仅供个人学习、研究或参考使用。若引用了第三方作品，版权归原作者所有。如内容涉及您的权益，请联系我们处理。