Python中的协程异常处理与传播机制

1. 问题背景

在异步编程中，协程（Coroutine）可能抛出异常，但异常的处理方式与同步代码不同。若未正确捕获异常，可能导致整个事件循环停止或其他协程被静默影响。理解协程异常的传播路径和捕获方法，是编写健壮异步代码的关键。

2. 协程异常的基本行为

（1）同步异常 vs 异步异常

• 同步异常：在协程函数内直接抛出的异常（如 raise ValueError），会立即中断当前协程的执行。
• 异步异常：通过事件循环或外部任务（Task）触发的异常（如 Task.cancel() 引发的 CancelledError）。

（2）示例：未捕获异常导致任务失败

import asyncio

async def faulty_coroutine():
    raise ValueError("协程内部错误")

async def main():
    task = asyncio.create_task(faulty_coroutine())
    await task  # 此处会抛出异常，导致main()协程终止

# 运行结果：ValueError: 协程内部错误
asyncio.run(main())

3. 协程异常的捕获方法

（1）使用 try...except 包裹 await

在调用协程时，通过 await 直接等待其完成，可像同步代码一样捕获异常：

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

（2）通过 Task 对象获取异常

若协程被包装为 Task 并并发执行，异常不会立即抛出，而是存储在 Task 对象中：

async def main():
    task = asyncio.create_task(faulty_coroutine())
    await asyncio.sleep(0.1)  # 给任务执行时间
    if task.done() and not task.cancelled():
        exc = task.exception()  # 获取异常对象（若无异常返回None）
        if exc:
            print(f"任务异常: {exc}")

4. 异常在协程链中的传播

（1）异常会向上层调用者传播

async def child():
    raise ValueError("子协程异常")

async def parent():
    await child()  # 异常会从child传播到parent

async def grandparent():
    try:
        await parent()
    except ValueError as e:
        print(f"祖协程捕获: {e}")

asyncio.run(grandparent())  # 输出：祖协程捕获: 子协程异常

（2）关键规则：

• 若协程中未处理异常，异常会传递给 await 它的调用者。
• 若异常未被任何协程捕获，最终会传递给事件循环，导致程序终止。

5. 特殊异常：CancelledError

（1）取消任务时的行为

当调用 Task.cancel() 时，任务会在下一个 await 点抛出 CancelledError：

async def cancel_me():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("任务被取消！")
        raise  # 必须重新抛出，否则任务状态不会更新

async def main():
    task = asyncio.create_task(cancel_me())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("主协程确认任务已取消")

asyncio.run(main())

（2）注意：

• 捕获 CancelledError 后需决定是否重新抛出：若静默处理，任务可能无法正常取消。

6. 使用 asyncio.gather 的异常处理

（1）默认行为：任一协程异常立即传播

async def coro1():
    await asyncio.sleep(1)

async def coro2():
    raise ValueError("coro2失败")

async def main():
    try:
        await asyncio.gather(coro1(), coro2())
    except Exception as e:
        print(f"捕获异常: {e}")  # 仅看到coro2的异常，coro1被取消

（2）设置 return_exceptions=True

将异常作为结果返回，而非直接抛出：

async def main():
    results = await asyncio.gather(coro1(), coro2(), return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"协程{i}异常: {r}")
        else:
            print(f"协程{i}成功: {r}")

7. 最佳实践与总结

1. 始终在顶层协程捕获异常：例如在 main() 中使用 try...except 包裹事件循环。
2. 谨慎处理 CancelledError：确保关键清理操作执行后重新抛出。
3. 使用 Task.exception() 检查并发任务状态：避免异常被静默忽略。
4. 利用 gather 的 return_exceptions：当需要处理部分失败场景时。

通过以上步骤，你可以系统掌握协程异常的传播路径和处理策略，避免异步程序因未处理异常而崩溃。