错误发生时显示Python logging最新日志消息的实现方案
Python logging实现错误触发时查看模块最新日志的方案
需求背景
日常运行时将checker模块的日志级别设为ERROR,避免大量追踪日志占用资源;但当调用checker的runner模块发生错误时,需要能查看checker模块的最新日志记录(类似C++ spdlog库的环形缓冲区回溯功能)。
实现方案
方案1:利用MemoryHandler实现条件触发的日志缓存
logging.handlers.MemoryHandler是标准库内置的内存缓存Handler,可临时存储指定数量的日志记录,当触发条件满足时(如捕获到ERROR级别的日志),将缓存的日志转发到目标Handler输出。
示例代码:
import logging from logging.handlers import MemoryHandler # 1. 配置最终输出用的Handler console_handler = logging.StreamHandler() console_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') console_handler.setFormatter(console_formatter) # 2. 配置MemoryHandler:缓存最近50条checker日志,当出现ERROR级日志时触发输出 memory_handler = MemoryHandler( capacity=50, flushLevel=logging.ERROR, target=console_handler ) # 3. 配置checker模块的logger:记录所有级别日志,但仅用MemoryHandler缓存 checker_logger = logging.getLogger('checker') checker_logger.setLevel(logging.DEBUG) checker_logger.addHandler(memory_handler) # 4. 配置runner模块的logger:仅输出ERROR级日志 runner_logger = logging.getLogger('runner') runner_logger.setLevel(logging.ERROR) runner_logger.addHandler(console_handler) # 模拟业务场景 def run_checker(): # 生成大量日常追踪日志(平时不会输出) for i in range(100): checker_logger.debug(f'checker追踪信息 {i}') # 模拟runner调用checker后出错 runner_logger.error('runner执行失败') # 触发MemoryHandler输出缓存的checker日志
说明:当runner模块输出ERROR级别日志时,MemoryHandler会自动将缓存的checker模块最新50条日志(包括DEBUG级追踪信息)全部输出,满足错误时回溯的需求。
方案2:自定义环形缓冲区Handler(更贴近spdlog的backtrace)
如果需要严格的环形缓冲区行为(固定大小,始终保留最新N条记录,满了自动丢弃最旧的),可以自定义Handler:
import logging from collections import deque class RingBufferHandler(logging.Handler): def __init__(self, capacity=50): super().__init__() self.buffer = deque(maxlen=capacity) # 固定长度队列自动实现环形缓存 def emit(self, record): self.buffer.append(self.format(record)) def flush_buffer(self, target_handler): # 将缓存日志输出到指定Handler for msg in self.buffer: log_record = logging.makeLogRecord({'msg': msg}) target_handler.emit(log_record) # 使用示例 console_handler = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') console_handler.setFormatter(formatter) # 初始化环形缓冲区Handler,缓存50条日志 ring_handler = RingBufferHandler(capacity=50) ring_handler.setFormatter(formatter) # 配置checker模块logger checker_logger = logging.getLogger('checker') checker_logger.setLevel(logging.DEBUG) checker_logger.addHandler(ring_handler) # 配置runner模块logger runner_logger = logging.getLogger('runner') runner_logger.setLevel(logging.ERROR) # 模拟错误场景 def simulate_error(): # 生成大量checker追踪日志 for i in range(100): checker_logger.debug(f'checker追踪日志 {i}') # runner出错时手动输出缓存的checker日志 try: raise ValueError('runner执行异常') except Exception as e: runner_logger.error(f'runner错误: {e}') print('\n--- checker模块最新日志回溯 ---') ring_handler.flush_buffer(console_handler)
说明:自定义的RingBufferHandler用deque实现环形缓存,当runner模块出错时,手动调用flush_buffer方法即可输出缓存的最新日志,完全匹配spdlog的backtrace逻辑。
Python logging与spdlog回溯功能的对比
Python标准库logging没有内置类似spdlog的一键回溯功能,但通过上述两种方案可以实现等价效果:
- 利用
MemoryHandler可实现自动触发的日志缓存与输出 - 自定义环形缓冲区Handler能更灵活地控制缓存行为,完全复刻spdlog的环形缓冲区回溯逻辑
内容的提问来源于stack exchange,提问作者morxa
相关产品推荐
相关产品推荐

