如何编写捕获协程全局异常错误脚本

本文目录导读：

1. Python asyncio 全局异常捕获
2. 完善的全局异常捕获框架
3. 使用 asyncio 的 await 和 Task 异常
4. 使用建议

Python asyncio 全局异常捕获

使用 loop.set_exception_handler()

import asyncio
import sys
import traceback
def global_exception_handler(loop, context):
    """
    全局异常处理器
    """
    # 获取异常信息
    exception = context.get('exception')
    message = context.get('message', '发生未知错误')
    # 获取调用栈信息
    if exception:
        print(f"全局异常捕获: {type(exception).__name__}: {exception}")
        traceback.print_exception(type(exception), exception, exception.__traceback__)
    else:
        print(f"错误信息: {message}")
    # 获取出错的协程
    future = context.get('future')
    if future:
        print(f"出错的协程: {future}")
    # 可以记录到日志文件
    with open('error.log', 'a') as f:
        f.write(f"{message}\n")
        if exception:
            traceback.print_exception(type(exception), exception, exception.__traceback__, file=f)
async def risky_task(name):
    """可能抛出异常的任务"""
    await asyncio.sleep(1)
    if name == "task2":
        raise ValueError(f"任务 {name} 出错!")
    print(f"任务 {name} 完成")
    return name
async def main():
    # 获取事件循环并设置全局异常处理器
    loop = asyncio.get_event_loop()
    loop.set_exception_handler(global_exception_handler)
    # 创建多个任务
    tasks = [
        asyncio.create_task(risky_task("task1")),
        asyncio.create_task(risky_task("task2")),
        asyncio.create_task(risky_task("task3")),
    ]
    # 等待所有任务完成（包括出错的任务）
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # 处理结果
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"任务 {i+1} 返回异常: {result}")
        else:
            print(f"任务 {i+1} 返回结果: {result}")
if __name__ == "__main__":
    asyncio.run(main())

使用装饰器捕获单个协程异常

import asyncio
import functools
import traceback
def catch_coroutine_error(func):
    """
    装饰器：捕获协程中的异常
    """
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            print(f"协程 {func.__name__} 发生异常: {e}")
            traceback.print_exc()
            # 可以选择重新抛出、返回默认值或记录日志
            return None  # 返回默认值
    return wrapper
@catch_coroutine_error
async def risky_operation(name):
    """可能出错的操作"""
    await asyncio.sleep(1)
    raise RuntimeError(f"{name} 操作失败")
    return "成功"
async def main():
    results = await asyncio.gather(
        risky_operation("操作1"),
        risky_operation("操作2"),
        return_exceptions=False
    )
    print(f"结果: {results}")
asyncio.run(main())

完善的全局异常捕获框架

import asyncio
import logging
import traceback
import sys
from typing import Optional, Callable, Any
class GlobalExceptionHandler:
    """全局异常捕获管理器"""
    def __init__(self):
        self.setup_logging()
        self.handlers = []
    def setup_logging(self):
        """设置日志系统"""
        logging.basicConfig(
            level=logging.ERROR,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('async_errors.log'),
                logging.StreamHandler(sys.stdout)
            ]
        )
        self.logger = logging.getLogger(__name__)
    def add_handler(self, handler: Callable):
        """添加自定义异常处理器"""
        self.handlers.append(handler)
    def exception_handler(self, loop: asyncio.AbstractEventLoop, context: dict):
        """全局异常处理器"""
        exception = context.get('exception')
        message = context.get('message', '未捕获的异常')
        # 记录日志
        self.logger.error(f"全局异常: {message}")
        if exception:
            self.logger.error(f"异常类型: {type(exception).__name__}")
            self.logger.error(f"异常信息: {str(exception)}")
            self.logger.error(f"堆栈信息:\n{''.join(traceback.format_tb(exception.__traceback__))}")
        # 调用自定义处理器
        for handler in self.handlers:
            try:
                handler(loop, context)
            except Exception as e:
                self.logger.error(f"自定义处理器出错: {e}")
        # 可以在这里决定是否终止程序
        # loop.stop()
class SafeTask:
    """安全任务包装器"""
    def __init__(self, coro, name: str = None, on_error: Optional[Callable] = None):
        self.coro = coro
        self.name = name or getattr(coro, '__name__', 'unknown')
        self.on_error = on_error
    async def run(self):
        """安全执行协程"""
        try:
            return await self.coro
        except asyncio.CancelledError:
            logging.warning(f"任务 {self.name} 被取消")
            raise
        except Exception as e:
            error_info = {
                'task_name': self.name,
                'exception': e,
                'traceback': traceback.format_exc()
            }
            # 记录错误
            logging.error(f"任务 {self.name} 执行失败: {e}")
            logging.error(f"完整错误信息:\n{traceback.format_exc()}")
            # 调用自定义错误处理函数
            if self.on_error:
                await self.on_error(error_info)
            # 可以选择重新抛出或返回默认值
            return None
async def example_usage():
    """使用示例"""
    # 创建全局异常处理器
    error_handler = GlobalExceptionHandler()
    # 添加自定义处理器
    async def custom_handler(loop, context):
        print(f"自定义处理: {context.get('message')}")
    error_handler.add_handler(custom_handler)
    # 设置事件循环的异常处理器
    loop = asyncio.get_event_loop()
    loop.set_exception_handler(error_handler.exception_handler)
    # 使用安全包装器执行任务
    async def risky_task():
        raise ValueError("模拟错误")
    safe_task = SafeTask(
        risky_task(), 
        name="risky_task",
        on_error=lambda info: print(f"任务错误回调: {info['exception']}")
    )
    result = await safe_task.run()
    print(f"安全任务返回: {result}")
    # 或者直接创建任务
    task = asyncio.create_task(risky_task())
    try:
        await task
    except Exception as e:
        print(f"捕获到异常: {e}")
if __name__ == "__main__":
    asyncio.run(example_usage())

使用 asyncio 的 await 和 Task 异常

import asyncio
import traceback
async def task_with_timeout():
    """带超时的任务"""
    try:
        await asyncio.sleep(2)
        raise ValueError("任务超时异常")
    except asyncio.TimeoutError:
        print("任务超时")
        return None
async def monitor_tasks(tasks):
    """监控任务执行情况"""
    for task in tasks:
        if task.done():
            if not task.cancelled() and task.exception():
                print(f"任务异常: {task.exception()}")
                traceback.print_exception(
                    type(task.exception()), 
                    task.exception(), 
                    task.exception().__traceback__
                )
async def main():
    # 创建并启动任务
    tasks = [
        asyncio.create_task(task_with_timeout(), name=f"task_{i}")
        for i in range(3)
    ]
    # 定期检查任务状态
    while not all(t.done() for t in tasks):
        await asyncio.sleep(0.1)
        monitor_tasks(tasks)
    # 收集结果
    for task in tasks:
        try:
            result = await task
            print(f"任务 {task.get_name()} 完成: {result}")
        except Exception as e:
            print(f"任务 {task.get_name()} 异常: {e}")
asyncio.run(main())

使用建议

1. 选择合适的方案：

   • 全局处理器：统一处理所有未捕获异常
   • 装饰器：精确控制单个协程
   • 安全包装器：灵活的异常处理

2. 日志记录：务必记录异常日志便于调试

3. 异常恢复：考虑是否允许程序继续运行

4. 资源清理：异常发生时确保释放资源

5. 测试覆盖：确保异常处理逻辑经过充分测试

这些方案可以根据项目需求灵活组合使用。