Python 异常处理在异步编程中的写法

冷漠man 2025-11-20 00:00:00 次阅读

答案：在Python异步编程中，异常需通过await触发并用try-except捕获；使用asyncio.gather时可设return_exceptions=True避免中断；Task异常需显式await或result()触发；超时场景需捕获TimeoutError，关键在于正确处理await层级以防遗漏异常。

在 Python 异步编程中，异常处理与同步代码类似，但需注意协程的执行方式和上下文。使用 try-except 块可以捕获异步函数中的异常，但必须在 await 表达式上进行捕获，因为异常通常是在等待协程完成时抛出的。

基本异常捕获（单个 await）

在 async 函数中调用另一个协程时，如果该协程抛出异常，需要在 try 中 await 它：

import asyncio

async def might_fail():
    await asyncio.sleep(1)
    raise ValueError("出错了！")

async def main():
    try:
        await might_fail()
    except ValueError as e:
        print(f"捕获到异常: {e}")

asyncio.run(main())

并发任务中的异常处理（gather）

asyncio.gather 可以并发运行多个协程，并集中处理异常。默认情况下，只要有一个协程出错，gather 就会抛出异常。可以通过设置 return_exceptions=True 来避免中断，让所有任务继续执行。

async def task1():
    raise RuntimeError("任务1失败")

async def task2():
    return "任务2成功"

async def task3():
    return "任务3成功"

async def main():
    results = await asyncio.gather(
        task1(), task2(), task3(),
        return_exceptions=True
    )
    
    for result in results:
        if isinstance(result, Exception):
            print(f"任务出错: {result}")
        else:
            print(f"任务结果: {result}")

asyncio.run(main())

使用 Task 和 wait 处理异常

通过创建 Task 并使用 asyncio.wait，可以更灵活地控制异常处理时机。任务即使出错也不会立即抛出，而是等到你显式调用 result() 或 await 时才触发。

async def bad_task():
    await asyncio.sleep(1)
    raise KeyError("键错误")

async def main():
    task = asyncio.create_task(bad_task())
    
    try:
        await task
    except KeyError as e:
        print(f"捕获任务异常: {e}")

异常在回调或超时场景中的处理

使用 asyncio.wait_for 设置超时时，可能会抛出 TimeoutError，也需要捕获：

async def slow_function():
    await asyncio.sleep(10)

async def main():
    try:
        await asyncio.wait_for(slow_function(), timeout=3)
    except asyncio.TimeoutError:
        print("函数执行超时")

基本上就这些。关键是理解：异步异常不会自动传播，必须通过 await 触发并用 try-except 捕获。并发场景推荐使用 gather + return_exceptions 或监控 Task 状态来确保不遗漏错误。写法不复杂，但容易忽略 await 层级导致异常未被捕获。