Python脚本如何捕获协程异常报错

本文目录导读：

1. 使用 try/except 捕获协程函数内的异常
2. 在调用处捕获异常
3. 使用 asyncio.gather() 捕获多个协程的异常
4. 使用 asyncio.create_task() 捕获任务异常
5. 使用 Task.exception() 检查异常
6. 全局异常处理（自定义 Future）
7. 实用示例：完整的错误处理模式
8. 最佳实践建议：

在 Python 中捕获协程（coroutine）异常主要有以下几种方式：

使用 try/except 捕获协程函数内的异常

import asyncio
async def my_coroutine():
    try:
        # 可能出错的代码
        result = 1 / 0
        return result
    except ZeroDivisionError as e:
        print(f"捕获到异常: {e}")
        return None
async def main():
    await my_coroutine()
asyncio.run(main())

在调用处捕获异常

import asyncio
async def risky_coroutine():
    # 可能会抛出异常
    return 1 / 0
async def main():
    try:
        await risky_coroutine()
    except ZeroDivisionError as e:
        print(f"在调用处捕获异常: {e}")
asyncio.run(main())

使用 asyncio.gather() 捕获多个协程的异常

import asyncio
async def coro1():
    raise ValueError("错误1")
async def coro2():
    return "成功"
async def main():
    results = await asyncio.gather(
        coro1(), 
        coro2(), 
        return_exceptions=True  # 返回异常对象而不是抛出
    )
    for result in results:
        if isinstance(result, Exception):
            print(f"捕获到异常: {result}")
        else:
            print(f"成功结果: {result}")
asyncio.run(main())

使用 asyncio.create_task() 捕获任务异常

import asyncio
async def risky_task():
    await asyncio.sleep(1)
    raise RuntimeError("任务出错")
async def main():
    task = asyncio.create_task(risky_task())
    try:
        await task
    except RuntimeError as e:
        print(f"任务异常: {e}")
asyncio.run(main())

使用 Task.exception() 检查异常

import asyncio
async def risky_task():
    await asyncio.sleep(1)
    raise ValueError("出错了")
async def main():
    task = asyncio.create_task(risky_task())
    # 等待任务完成
    await asyncio.sleep(2)
    if task.done() and not task.cancelled():
        try:
            task.result()  # 如果任务有异常，这里会抛出
        except ValueError as e:
            print(f"检查到任务异常: {e}")
asyncio.run(main())

全局异常处理（自定义 Future）

import asyncio
import sys
async def risky_coroutine():
    raise Exception("未知错误")
def handle_exception(loop, context):
    msg = context.get("exception", context["message"])
    print(f"全局异常: {msg}")
async def main():
    loop = asyncio.get_running_loop()
    loop.set_exception_handler(handle_exception)
    # 这个异常会被全局处理器捕获
    await risky_coroutine()
asyncio.run(main())

实用示例：完整的错误处理模式

import asyncio
import logging
logging.basicConfig(level=logging.INFO)
async def safe_coroutine(coro, timeout=5):
    """安全执行协程的封装"""
    try:
        return await asyncio.wait_for(coro, timeout)
    except asyncio.TimeoutError:
        logging.error(f"协程执行超时")
        return None
    except Exception as e:
        logging.error(f"协程执行出错: {e}", exc_info=True)
        return None
async def example_with_error():
    raise ConnectionError("连接失败")
async def main():
    # 安全的执行协程
    result = await safe_coroutine(example_with_error())
    if result is None:
        print("协程执行失败，继续处理其他逻辑")
asyncio.run(main())

最佳实践建议：

1. 尽量在协程内部捕获异常，保持代码清晰
2. 使用 return_exceptions=True 在 gather 中处理多个协程
3. 避免使用裸 except，指定具体的异常类型
4. 使用日志记录异常信息，便于调试
5. 考虑使用 asyncio.shield() 保护关键操作不被取消

选择哪种方式取决于你的具体需求,但通常推荐在调用处使用 try/except 进行捕获，这样可以保持异常的传播清晰可控。