Python日志系统详解：从基础配置到高级应用

1. Python日志系统概述logging模块是Python标准库中用于应用程序和库的灵活事件日志系统。它提供了一套完整的日志记录功能允许开发者根据不同的需求对日志进行分级、过滤和输出。日志系统的基本组件包括Logger记录器应用程序代码直接交互的接口Handler处理器决定日志输出的位置控制台、文件等Filter过滤器提供更细粒度的日志过滤功能Formatter格式化器控制日志的最终输出格式2. 日志级别详解Python logging模块定义了6个标准日志级别按严重性递增排列2.1 标准日志级别NOTSET (0)表示不设置级别会继承父logger的级别DEBUG (10)详细的调试信息通常只在开发时使用INFO (20)确认程序按预期运行的一般信息WARNING (30)表示发生了意外情况或即将发生问题ERROR (40)由于更严重的问题程序无法执行某些功能CRITICAL (50)严重错误程序本身可能无法继续运行提示在实际应用中建议根据环境设置不同的日志级别。开发环境可以使用DEBUG生产环境建议使用INFO或WARNING级别。2.2 日志级别设置方法设置日志级别有两种方式# 方式1使用数值 logger.setLevel(10) # 设置为DEBUG级别 # 方式2使用字符串推荐 logger.setLevel(DEBUG)3. Logger对象深入解析Logger是日志系统的核心组件每个Logger都有一个名称形成层次结构。3.1 Logger的层次结构Logger名称使用点号分隔如a、a.b、a.b.c。未命名的logger会返回root logger。import logging # 获取root logger root_logger logging.getLogger() # 获取名为app的logger app_logger logging.getLogger(app) # 获取名为app.module的logger是app的子logger module_logger logging.getLogger(app.module)3.2 Logger的继承机制子logger会继承父logger的日志级别和handler除非显式设置# 设置父logger级别 logging.getLogger(app).setLevel(INFO) # 子logger未设置级别会继承父logger的INFO级别 module_logger logging.getLogger(app.module) print(module_logger.getEffectiveLevel()) # 输出20INFO3.3 Logger的常用方法日志记录方法debug(msg, *args, **kwargs)info(msg, *args, **kwargs)warning(msg, *args, **kwargs)error(msg, *args, **kwargs)critical(msg, *args, **kwargs)log(lvl, msg, *args, **kwargs)配置方法setLevel(lvl)addHandler(hdlr)removeHandler(hdlr)addFilter(filter)removeFilter(filter)4. Handler处理器详解Handler决定日志的输出位置和方式。常用的内置Handler包括4.1 常用Handler类型StreamHandler输出到流如sys.stderrFileHandler输出到文件RotatingFileHandler按大小自动分割日志文件TimedRotatingFileHandler按时间自动分割日志文件SMTPHandler通过邮件发送日志SocketHandler通过TCP发送日志4.2 Handler配置示例import logging # 创建logger logger logging.getLogger(example) logger.setLevel(logging.DEBUG) # 创建控制台handler并设置级别 console_handler logging.StreamHandler() console_handler.setLevel(logging.DEBUG) # 创建文件handler并设置级别 file_handler logging.FileHandler(example.log) file_handler.setLevel(logging.WARNING) # 创建formatter并添加到handler formatter logging.Formatter(%(asctime)s - %(name)s - %(levelname)s - %(message)s) console_handler.setFormatter(formatter) file_handler.setFormatter(formatter) # 将handler添加到logger logger.addHandler(console_handler) logger.addHandler(file_handler) # 记录日志 logger.debug(Debug message) logger.info(Info message) logger.warning(Warning message) logger.error(Error message) logger.critical(Critical message)5. Formatter格式化器Formatter控制日志的最终输出格式支持丰富的属性替换。5.1 常用格式化属性属性名格式描述asctime%(asctime)s易读的时间格式created%(created)f时间戳time.time()返回值filename%(filename)s文件名部分funcName%(funcName)s包含日志调用的函数名levelname%(levelname)s日志级别文本DEBUG, INFO等levelno%(levelno)s日志级别数字lineno%(lineno)d源代码行号message%(message)s记录的消息module%(module)s模块名name%(name)slogger名称pathname%(pathname)s完整路径名process%(process)d进程IDprocessName%(processName)s进程名thread%(thread)d线程IDthreadName%(threadName)s线程名5.2 自定义Formatter示例# 创建详细的formatter detailed_formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(module)s.%(funcName)s:%(lineno)d - %(message)s, datefmt%Y-%m-%d %H:%M:%S ) # 创建简洁的formatter simple_formatter logging.Formatter(%(levelname)s: %(message)s) # 应用到不同的handler console_handler.setFormatter(simple_formatter) file_handler.setFormatter(detailed_formatter)6. 日志配置最佳实践6.1 基础配置方法最简单的配置方式是使用basicConfig()import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, filenameapp.log, filemodea )6.2 从配置文件加载配置更复杂的配置可以从字典或文件加载import logging.config # 配置字典 config { version: 1, formatters: { detailed: { format: %(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s } }, handlers: { console: { class: logging.StreamHandler, level: INFO, formatter: detailed }, file: { class: logging.FileHandler, filename: mplog.log, level: DEBUG, formatter: detailed } }, root: { level: DEBUG, handlers: [console, file] } } logging.config.dictConfig(config)6.3 多模块日志实践在大型项目中建议在主模块中配置日志系统在各子模块中获取logger时使用__name__作为名称# 主模块 main.py import logging import my_module logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) logger.info(Starting application) my_module.do_something() logger.info(Finished) # 子模块 my_module.py import logging logger logging.getLogger(__name__) def do_something(): logger.info(Doing something)7. 高级日志技巧7.1 上下文信息添加可以通过extra参数添加上下文信息logger.info(User logged in, extra{ip: 192.168.1.1, user: admin})需要相应的formatter支持formatter logging.Formatter(%(asctime)s %(ip)-15s %(user)-8s %(message)s)7.2 异常日志记录记录异常信息的最佳实践try: 1 / 0 except ZeroDivisionError: logger.exception(Division by zero) # 会自动记录异常堆栈 # 等价于 # logger.error(Division by zero, exc_infoTrue)7.3 日志过滤自定义过滤器示例class InfoAndAboveFilter(logging.Filter): def filter(self, record): return record.levelno logging.INFO # 应用过滤器 handler.addFilter(InfoAndAboveFilter())7.4 日志轮转使用RotatingFileHandler实现日志轮转from logging.handlers import RotatingFileHandler handler RotatingFileHandler( app.log, maxBytes1024*1024, # 1MB backupCount5 # 保留5个备份 )8. 性能优化与线程安全8.1 日志性能优化避免在日志消息中进行字符串格式化# 不推荐即使日志级别高于DEBUG也会执行字符串格式化 logger.debug(Value: %s % expensive_function()) # 推荐使用%风格的格式化 logger.debug(Value: %s, expensive_function()) # 推荐使用{}或f-string风格的格式化 logger.debug(Value: {}.format(expensive_function()))使用isEnabledFor检查日志级别if logger.isEnabledFor(logging.DEBUG): logger.debug(Message with %s, expensive_data())8.2 线程安全logging模块是线程安全的但在信号处理函数中使用时需要注意import logging import signal # 不安全的信号处理函数 def handle_signal(signum, frame): logging.info(Received signal %s, signum) # 可能引发死锁 # 安全的替代方案 def handle_signal_safe(signum, frame): msg Received signal {}.format(signum) logging.Handler().emit(logging.LogRecord( name__name__, levellogging.INFO, pathname__file__, lineno0, msgmsg, argsNone, exc_infoNone )) signal.signal(signal.SIGINT, handle_signal_safe)9. 常见问题与解决方案9.1 日志重复输出问题原因多次添加handler导致 解决方案# 在添加handler前先移除现有handler logger.handlers.clear() logger.addHandler(handler) # 或者检查是否已有handler if not logger.handlers: logger.addHandler(handler)9.2 日志不输出问题检查步骤确认logger的级别设置确认handler的级别设置确认handler是否添加到logger确认formatter是否正确设置9.3 多进程日志问题在多进程环境下基本的FileHandler会导致日志混乱。解决方案from logging.handlers import QueueHandler, QueueListener import multiprocessing def worker_process(queue): # 所有日志通过QueueHandler发送到主进程 qh QueueHandler(queue) logger logging.getLogger() logger.addHandler(qh) logger.setLevel(logging.INFO) logger.info(Message from worker) def main(): queue multiprocessing.Queue() # 主进程设置QueueListener处理所有日志 handler logging.FileHandler(multiprocess.log) listener QueueListener(queue, handler) listener.start() processes [] for i in range(3): p multiprocessing.Process(targetworker_process, args(queue,)) processes.append(p) p.start() for p in processes: p.join() listener.stop() if __name__ __main__: main()10. 实际项目中的日志策略10.1 开发环境与生产环境配置建议根据环境使用不同的日志配置import logging import sys def setup_logging(envdev): logger logging.getLogger() logger.setLevel(logging.DEBUG) if env dev: # 开发环境详细日志输出到控制台 handler logging.StreamHandler(sys.stdout) formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) handler.setFormatter(formatter) else: # 生产环境关键日志输出到文件并按日轮转 from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( app.log, whenmidnight, backupCount7 ) formatter logging.Formatter( %(asctime)s - %(levelname)s - %(message)s ) handler.setFormatter(formatter) handler.setLevel(logging.INFO) logger.addHandler(handler)10.2 结构化日志对于需要日志分析的系统建议使用JSON格式import json import logging class JsonFormatter(logging.Formatter): def format(self, record): log_record { timestamp: self.formatTime(record), level: record.levelname, message: record.getMessage(), logger: record.name, module: record.module, function: record.funcName, line: record.lineno, } if record.exc_info: log_record[exception] self.formatException(record.exc_info) return json.dumps(log_record) # 使用示例 handler logging.StreamHandler() handler.setFormatter(JsonFormatter()) logger.addHandler(handler)10.3 审计日志对于需要记录用户操作的场景audit_logger logging.getLogger(audit) audit_logger.propagate False # 避免被root logger处理 audit_handler logging.FileHandler(audit.log) audit_formatter logging.Formatter( %(asctime)s %(user)s %(action)s %(status)s %(details)s ) audit_handler.setFormatter(audit_formatter) audit_logger.addHandler(audit_handler) # 记录审计日志 def log_audit_event(user, action, status, detailsNone): extra { user: user, action: action, status: status, details: details or } audit_logger.info(, extraextra)11. 测试中的日志处理11.1 单元测试中的日志验证可以使用MemoryHandler捕获日志进行验证import logging import unittest from io import StringIO class TestLogging(unittest.TestCase): def setUp(self): self.log_stream StringIO() handler logging.StreamHandler(self.log_stream) handler.setFormatter(logging.Formatter(%(levelname)s:%(message)s)) self.logger logging.getLogger(test) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO) def test_log_output(self): self.logger.info(Test message) log_content self.log_stream.getvalue() self.assertIn(INFO:Test message, log_content)11.2 模拟日志记录使用unittest.mock可以模拟日志记录from unittest.mock import patch class TestWithMockLogging(unittest.TestCase): patch(logging.getLogger) def test_logging(self, mock_get_logger): mock_logger mock_get_logger.return_value # 测试代码 some_function_that_logs() # 验证日志调用 mock_logger.info.assert_called_with(Expected message)12. 与其他日志系统集成12.1 与系统日志集成在Linux系统上可以输出到syslogfrom logging.handlers import SysLogHandler syslog_handler SysLogHandler(address/dev/log) syslog_handler.setFormatter(logging.Formatter(%(name)s: %(message)s)) logger.addHandler(syslog_handler)12.2 与日志收集系统集成对于ELK、Splunk等日志收集系统可以使用相应的Handler# 示例使用HTTPHandler发送日志到Logstash from logging.handlers import HTTPHandler http_handler HTTPHandler( logstash.example.com:8080, /log, methodPOST ) logger.addHandler(http_handler)13. 自定义日志级别虽然不推荐但可以添加自定义日志级别TRACE 5 logging.addLevelName(TRACE, TRACE) def trace(self, message, *args, **kwargs): if self.isEnabledFor(TRACE): self._log(TRACE, message, args, **kwargs) logging.Logger.trace trace # 使用 logger.setLevel(TRACE) logger.trace(Very detailed trace message)14. 性能关键代码的日志优化对于性能敏感的代码可以使用延迟日志记录class LazyMessage: def __init__(self, func, *args, **kwargs): self.func func self.args args self.kwargs kwargs def __str__(self): return str(self.func(*self.args, **self.kwargs)) def expensive_operation(): # 模拟耗时操作 import time time.sleep(1) return Result # 只有在日志级别足够时才会调用expensive_operation logger.debug(Result: %s, LazyMessage(expensive_operation))15. 日志文件管理策略15.1 按时间轮转from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( app.log, whenmidnight, # 每天轮转 interval1, backupCount7 # 保留7天 )15.2 按大小轮转from logging.handlers import RotatingFileHandler handler RotatingFileHandler( app.log, maxBytes10*1024*1024, # 10MB backupCount5 )15.3 压缩旧日志import gzip import os from logging.handlers import RotatingFileHandler class CompressedRotatingFileHandler(RotatingFileHandler): def doRollover(self): super().doRollover() # 压缩除当前日志外的文件 for i in range(1, self.backupCount 1): logname f{self.baseFilename}.{i} if os.path.exists(logname): with open(logname, rb) as f_in: with gzip.open(f{logname}.gz, wb) as f_out: f_out.writelines(f_in) os.remove(logname)