本文目录导读：
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())
使用建议
-
选择合适的方案：
- 全局处理器：统一处理所有未捕获异常
- 装饰器：精确控制单个协程
- 安全包装器：灵活的异常处理
-
日志记录：务必记录异常日志便于调试
-
异常恢复：考虑是否允许程序继续运行
-
资源清理：异常发生时确保释放资源
-
测试覆盖：确保异常处理逻辑经过充分测试
这些方案可以根据项目需求灵活组合使用。
