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

wen python案例 2026-06-27 238

本文目录导读：

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

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

在 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())