Python生产级日志封装完整解析_细节决定一切

logging等级try: 1 / 0 except Exception as e: logger.exception(计算错误) ERROR:test:计算错误 Traceback (most recent call last): File test.py, line 6, in module 1 / 0 ZeroDivisionError: division by zero 没有堆栈信息的话: ERROR:test:计算错误 生产环境调试全靠堆栈信息logger.exception()是定位问题的第一道防线。# 数据库操作 try: user db.query(SELECT * FROM users WHERE id ?, user_id) except Exception as e: logger.exception(f查询用户失败, user_id{user_id}) return None # API 调用 try: response requests.get(fhttps://api.example.com/user/{user_id}) except Exception as e: logger.exception(f调用用户API失败, user_id{user_id}) raise # 文件处理 try: with open(file_path, r) as f: data f.read() except Exception as e: logger.exception(f读取文件失败, path{file_path})一、代码全景展示import logging from logging.handlers import RotatingFileHandler import os def get_logger(name, level, log_file): 获取配置好的logger支持日志轮转 logger logging.getLogger(name) logger.setLevel(level) # 避免重复添加handler if logger.handlers: return logger # 确保日志目录存在 log_dir os.path.dirname(log_file) if log_dir and not os.path.exists(log_dir): os.makedirs(log_dir) # 文件处理器10MB轮转保留5个备份 file_handler RotatingFileHandler( log_file, maxBytes10*1024*1024, backupCount5, encodingutf-8 ) file_handler.setLevel(level) # 控制台处理器 console_handler logging.StreamHandler() console_handler.setLevel(level) # 格式化器 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) logger.addHandler(file_handler) logger.addHandler(console_handler) return logger二、逐行深度解析第1-3行导入必要的模块import logging # Python标准日志库 from logging.handlers import RotatingFileHandler # 日志轮转处理器 import os # 操作系统接口用于路径处理生产环境考量logging是Python官方日志库成熟稳定RotatingFileHandler解决单一日志文件过大问题os模块确保跨平台路径兼容性第5行函数定义def get_logger(name, level, log_file):参数说明参数类型说明示例namestr日志器名称通常用__name__user_servicelevelint日志级别logging.INFOlog_filestr日志文件路径logs/app.log第6行文档字符串获取配置好的logger支持日志轮转生产环境必须有清晰的文档说明便于团队协作。第7-8行创建并设置日志器 对象已存在就获取,不存在就创建新的,保证单对象 logger logging.getLogger(name) # 获取或创建logger实例 logger.setLevel(level) # 设置日志级别过滤器关键知识点getLogger()是工厂方法模式相同name返回同一个实例单例setLevel()设置全局过滤级别低于此级别的日志不会处理子文件夹会自动继承父文件夹的LEVEL,自己有用自己的,自己没有用父文件的第10-12行避免重复添加Handler核心防护 保障单对象➕单配置 这里的handlers是一个list,放着我们的 文件处理器 和 控制台处理器 # 避免重复添加handler if logger.handlers: # 检查是否已有处理器 return logger # 直接返回现有实例生产环境重要性# 错误示例没有防护的代码 get_logger(app, logging.INFO, app.log) get_logger(app, logging.INFO, app.log) # 结果同一个logger会有2个相同的handler日志会重复输出两次 # 正确示例有防护的代码 logger1 get_logger(app, logging.INFO, app.log) logger2 get_logger(app, logging.INFO, app.log) print(logger1 is logger2) # Truehandler只添加一次第14-18行自动创建日志目录log_dir os.path.dirname(log_file) # 提取目录路径 if log_dir and not os.path.exists(log_dir): # 目录不存在时 os.makedirs(log_dir) # 递归创建所有缺失的目录log_file logs/2024/01/app.log log_dir os.path.dirname(log_file) # 结果logs/2024/01 # 自动创建 logs/2024/01 目录结构 # 避免手动创建目录导致的 FileNotFoundError第20-24行文件处理器日志轮转核心file_handler RotatingFileHandler( log_file, # 日志文件路径 maxBytes10*1024*1024, # 10MB后轮转 backupCount5, # 保留5个备份 encodingutf-8 # UTF-8编码支持中文 ) file_handler.setLevel(level) # 设置处理器级别轮转机制演示 初始状态 app.log (5MB) 写入5MB后总10MB app.log → 轮转 → app.log.1 新建 app.log 继续写入达到10MB时 app.log → app.log.1 app.log.1 → app.log.2 ... 以此类推 最多保留5个备份 app.log (当前) app.log.1 (最新备份) app.log.2 app.log.3 app.log.4 app.log.5 (最旧备份) 核心是备份数 backupCount总文件数 backupCount 1 第26-28行控制台处理器console_handler logging.StreamHandler() # 输出到标准输出 console_handler.setLevel(level) # 设置级别生产环境用途开发调试时实时查看日志容器化部署Docker/K8s时通过stdout收集日志CI/CD流水线中查看构建日志第30-34行日志格式化器formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s ) file_handler.setFormatter(formatter) # 文件使用完整格式 console_handler.setFormatter(formatter) # 控制台也使用相同格式格式占位符详解占位符说明示例输出%(asctime)s时间戳2024-01-15 10:30:45,123%(name)s日志器名称user_service%(levelname)s日志级别INFO/ERROR%(filename)s源文件名user_controller.py%(lineno)d行号42%(message)s日志消息用户登录成功输出示例 2024-01-15 10:30:45,123 - user_service - INFO - user_controller.py:42 - 用户登录成功 第36-37行组装并返回logger.addHandler(file_handler) # 添加文件处理器 logger.addHandler(console_handler) # 添加控制台处理器 return logger # 返回配置好的实例三、完整使用示例示例1基础使用# 初始化日志器 logger get_logger( namemy_app, levellogging.INFO, log_filelogs/my_app.log ) # 使用不同级别记录日志 logger.debug(这是调试信息) # 不会输出INFO级别高于DEBUG logger.info(应用启动成功) # ✅ 会输出 logger.warning(配置文件未找到使用默认配置) # ✅ 会输出 logger.error(数据库连接失败) # ✅ 会输出 logger.critical(系统崩溃) # ✅ 会输出 # 异常信息记录 try: 1 / 0 except Exception as e: logger.exception(计算错误) # 自动记录异常堆栈示例2多模块使用# database.py import logging from logger_config import get_logger logger get_logger(database, logging.INFO, logs/app.log) def connect_db(): logger.info(连接数据库...) # 数据库连接代码 # user_service.py from logger_config import get_logger logger get_logger(user_service, logging.INFO, logs/app.log) def create_user(username): logger.info(f创建用户: {username}) # 用户创建逻辑输出效果 2024-01-15 10:30:45,123 - database - INFO - database.py:5 - 连接数据库... 2024-01-15 10:30:46,456 - user_service - INFO - user_service.py:8 - 创建用户: admin 示例3Web框架集成Flaskfrom flask import Flask, request import logging app Flask(__name__) logger get_logger(web_app, logging.INFO, logs/web.log) app.before_request def log_request(): logger.info(f{request.method} {request.path} - IP: {request.remote_addr}) app.route(/api/user/int:uid) def get_user(uid): logger.info(f查询用户: {uid}) return {user_id: uid} if __name__ __main__: app.run()四、生产环境进阶优化优化1支持环境变量配置import os def get_logger_advanced(name): log_level getattr(logging, os.getenv(LOG_LEVEL, INFO)) log_file os.getenv(LOG_FILE, logs/app.log) return get_logger(name, log_level, log_file) # 使用环境变量控制 # LOG_LEVELDEBUG # LOG_FILE/var/log/myapp/app.log PyCharm 在点击运行时自动做了 source 的工作 PyCharm 自动加载了Linux 需要你手动 source .env linux中读取环境变量过程 # 1. 写 .env 文件 echo LOG_LEVELINFO .env # 2. 必须执行这个命令加载 source .env # 3. 然后才能运行 Python python app.py # ✅ 能读到 LOG_LEVEL 优化2不同模块不同日志文件# 业务日志 biz_logger get_logger(business, logging.INFO, logs/business.log) # 错误日志单独记录ERROR级别 error_logger get_logger(error, logging.ERROR, logs/error.log) # 访问日志 access_logger get_logger(access, logging.INFO, logs/access.log)优化3JSON格式输出适合日志收集系统import json import logging class JSONFormatter(logging.Formatter): def format(self, record): log_entry { timestamp: self.formatTime(record), level: record.levelname, logger: record.name, message: record.getMessage(), module: record.module, line: record.lineno } if record.exc_info: log_entry[exception] self.formatException(record.exc_info) return json.dumps(log_entry) # 使用JSON格式 formatter JSONFormatter() 输出: {timestamp: 2024-01-15 10:30:45,123, level: INFO, logger: app, message: 用户登录成功, module: test, line: 20} 五、常见问题与解决方案Q1日志重复输出原因多次调用get_logger导致重复添加handler解决代码中已有防护机制if logger.handlers: return loggerQ2中文乱码原因文件编码问题解决设置encodingutf-8Q3日志文件权限错误原因进程无写入权限解决# 创建目录时设置权限 os.makedirs(log_dir, mode0o755, exist_okTrue)Q4多进程写同一日志文件问题RotatingFileHandler不支持多进程解决方案from logging.handlers import QueueHandler, QueueListener from multiprocessing import Queue log_queue Queue(-1) queue_handler QueueHandler(log_queue) listener QueueListener(log_queue, file_handler, console_handler) listener.start()六、最佳实践总结实践要点说明✅ 使用__name__作为logger名称自动体现模块路径✅ 设置合理的轮转大小10-100MB为宜✅ 生产环境使用INFO级别DEBUG会影响性能✅ 异常使用logger.exception()自动记录堆栈信息✅ 避免在循环中记录大量日志影响性能✅ 敏感信息脱敏密码、token等不要记录七、性能对比测试import time # 测试1设置级别为CRITICAL logger.setLevel(logging.CRITICAL) start time.time() for i in range(100000): logger.debug(f测试消息 {i}) # 不输出但有判断开销 print(f禁用耗时: {time.time() - start:.3f}秒) # 测试2INFO级别 logger.setLevel(logging.INFO) start time.time() for i in range(100000): logger.debug(f测试消息 {i}) # 不输出但有判断开销 print(f判断开销: {time.time() - start:.3f}秒) # 测试3先判断再记录最佳实践 logger.setLevel(logging.INFO) start time.time() for i in range(100000): if logger.isEnabledFor(logging.DEBUG): logger.debug(f测试消息 {i}) print(f优化后: {time.time() - start:.3f}秒) logger.debug(f...) 无论什么级别f-string 都会先执行。 用 isEnabledFor 判断可以避免 f-string 的执行开销。 八、直接可用的完整模板 production_logger.py - 生产环境日志封装 import logging from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler import os from typing import Optional class ProductionLogger: 生产级日志管理器 _instances {} classmethod def get_logger(cls, name: str, log_level: Optional[int] None, log_file: Optional[str] None): 获取配置好的日志器 Args: name: 日志器名称建议使用__name__ log_level: 日志级别默认从环境变量LOG_LEVEL读取未设置则INFO log_file: 日志文件路径默认从环境变量LOG_FILE读取未设置则logs/app.log Returns: 配置好的Logger实例 # 配置默认值 if log_level is None: log_level getattr(logging, os.getenv(LOG_LEVEL, INFO)) if log_file is None: log_file os.getenv(LOG_FILE, logs/app.log) # 获取或创建logger logger logging.getLogger(name) # 避免重复配置 if name in cls._instances: return logger logger.setLevel(log_level) # 创建日志目录 log_dir os.path.dirname(log_file) if log_dir and not os.path.exists(log_dir): os.makedirs(log_dir, exist_okTrue) # 文件处理器按大小轮转 file_handler RotatingFileHandler( log_file, maxBytes10*1024*1024, # 10MB backupCount5, encodingutf-8 ) file_handler.setLevel(log_level) # 控制台处理器 console_handler logging.StreamHandler() console_handler.setLevel(log_level) # 格式化器 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s ) file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) # 添加处理器 logger.addHandler(file_handler) logger.addHandler(console_handler) # 缓存实例 cls._instances[name] logger return logger # 使用示例 if __name__ __main__: logger ProductionLogger.get_logger(__name__) logger.info(日志系统初始化完成) logger.warning(这是一条警告) try: result 10 / 0 except Exception as e: logger.exception(捕获到异常)测试执行结果: 2026-04-08 16:04:09,287 - __main__ - INFO - pp.py:84 - 日志系统初始化完成 2026-04-08 16:04:09,288 - __main__ - WARNING - pp.py:85 - 这是一条警告 2026-04-08 16:04:09,289 - __main__ - ERROR - pp.py:90 - 捕获到异常 Traceback (most recent call last): File D:\LLM\RAG\demo\d_tradtional_rag\utils\pp.py, line 88, in module result 10 / 0 ~~~^~~ ZeroDivisionError: division by zero 这份代码封装了生产环境日志的所有核心要素可直接用于实际项目。通过本文的逐行解析和示例相信你已经完全掌握了生产级日志系统的实现原理和最佳实践。