'# Python的logging模块（日志、DEBUG、INFO、WARNING、ERROR、CRITICAL）
一、背景与问题
在软件开发中，日志系统是调试、监控和故障排查的核心工具。Python的logging模块提供了灵活且功能强大的日志记录机制，但其复杂性常让开发者感到困惑。本文将深入解析logging模块的底层原理，探讨其在实际项目中的最佳实践，并通过完整案例展示其应用场景。
1.1 为什么需要日志系统？
- 调试：记录程序运行状态，定位错误
- 监控：跟踪系统行为，分析性能瓶颈
- 审计：记录关键操作，满足合规要求
- 故障恢复：快速定位问题根源
1.2 现有方案的局限性
简单print语句存在以下问题：
- 无法分级控制日志输出
- 难以管理日志文件生命周期
- 缺乏格式化能力
- 无法实现异步处理
二、基本原理
2.1 日志系统的层次结构
logging模块采用层次结构设计，包含三个核心组件：
Logger（日志记录器）
- 用于创建日志记录点
- 支持多级命名空间（
root、app、app.db等） - 可设置日志级别（DEBUG/INFO/WARNING/ERROR/CRITICAL）
Handler（处理器）
- 负责将日志消息发送到指定目的地（文件、控制台、网络等）
- 支持多种处理器类型（
StreamHandler、FileHandler、SMTPHandler等） - 可配置日志级别过滤
Formatter（格式器）
- 定义日志消息的格式
- 支持时间戳、日志级别、消息内容、文件名等字段
2.2 日志记录流程
- 使用
logger.info()等方法生成日志记录 - 日志记录器根据级别过滤后，将消息传递给所有注册的处理器
- 处理器根据配置将日志输出到指定目的地
- 格式器对日志消息进行格式化
三、环境准备
# 创建项目目录结构
mkdir logging_demo
cd logging_demo
mkdir src tests四、核心实现
4.1 基础日志记录
import logging
# 配置日志系统
logging.basicConfig(
level=logging.DEBUG, # 设置全局日志级别
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename='app.log', # 输出到文件
filemode='w' # 覆盖写入
)
# 创建日志记录器
logger = logging.getLogger(__name__)
# 记录不同级别的日志
logger.debug("调试信息")
logger.info("正常信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")关键代码解释：
level=logging.DEBUG：设置全局日志级别，低于该级别的日志不会被记录filename='app.log'：日志输出到文件，filemode='w'表示覆盖写入%(asctime)s：时间戳格式化字段%(name)s：日志记录器名称%(levelname)s：日志级别名称
4.2 自定义日志配置
import logging
# 创建日志记录器
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
# 创建控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 创建文件处理器
file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)
# 创建格式器
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# 绑定格式器
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
# 添加处理器
logger.addHandler(console_handler)
logger.addHandler(file_handler)
# 记录日志
logger.debug("调试信息")
logger.info("正常信息")
logger.warning("警告信息")
logger.error("错误信息")
logger.critical("严重错误")关键代码解释：
setLevel()方法设置处理器的日志级别，实现更细粒度控制StreamHandler将日志输出到控制台，FileHandler输出到文件- 通过
setFormatter()方法统一设置格式器
4.3 日志记录器层次结构
import logging
# 创建父记录器
parent_logger = logging.getLogger('root')
parent_logger.setLevel(logging.WARNING)
# 创建子记录器
child_logger = logging.getLogger('root.child')
child_logger.setLevel(logging.DEBUG)
# 记录日志
parent_logger.debug("父记录器调试信息") # 不会输出
parent_logger.info("父记录器信息") # 会输出
child_logger.debug("子记录器调试信息") # 会输出
child_logger.info("子记录器信息") # 会输出关键点：
- 父记录器的配置会影响子记录器
- 可通过
logging.getLogger(__name__)创建命名空间
五、完整案例
5.1 电商系统日志案例
# src/main.py
import logging
import os
from datetime import datetime
# 配置日志系统
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
filename=f'logs/{datetime.now().strftime("%Y%m%d")}.log',
filemode='a'
)
logger = logging.getLogger(__name__)
class OrderProcessor:
def __init__(self):
self.logger = logging.getLogger('order_processor')
self.logger.setLevel(logging.DEBUG)
self.logger.addHandler(logging.StreamHandler()) # 实时输出到控制台
def process_order(self, order_id):
logger.info(f"开始处理订单 {order_id}")
try:
self.validate_order(order_id)
self.calculate_price(order_id)
self.save_to_database(order_id)
except Exception as e:
logger.error(f"处理订单 {order_id} 出错: {str(e)}", exc_info=True)
raise
def validate_order(self, order_id):
logger.debug(f"验证订单 {order_id}")
if order_id % 2 == 0:
raise ValueError("无效订单ID")
def calculate_price(self, order_id):
logger.debug(f"计算订单 {order_id} 价格")
# 模拟计算过程
if order_id % 3 == 0:
raise RuntimeError("计算失败")
def save_to_database(self, order_id):
logger.debug(f"保存订单 {order_id} 到数据库")
# 模拟数据库保存
if order_id % 5 == 0:
raise ConnectionError("数据库连接失败")
# 调用示例
if __name__ == "__main__":
processor = OrderProcessor()
try:
processor.process_order(10)
except Exception as e:
logger.error(f"处理订单失败: {str(e)}")案例说明：
- 使用多级日志记录器跟踪订单处理流程
- 在异常处理中输出堆栈信息
- 日志文件按日期轮转
- 控制台实时输出调试信息
六、源码解析
6.1 日志记录器源码
class Logger:
def __init__(self, name):
self.name = name
self.handlers = []
self.level = logging.NOTSET # 默认级别
def setLevel(self, level):
self.level = level
def addHandler(self, handler):
self.handlers.append(handler)
def log(self, level, msg, *args, **kwargs):
if self.level <= level:
for handler in self.handlers:
handler.emit(msg)关键点：
- 日志记录器维护一个处理器列表
- 通过
setLevel()控制日志级别 log()方法实现日志记录逻辑
6.2 处理器源码
class Handler:
def __init__(self, level=logging.NOTSET):
self.level = level
def setFormatter(self, formatter):
self.formatter = formatter
def emit(self, record):
if self.level <= record.levelno:
self.format(record)
self.do_emit(record)
def format(self, record):
if self.formatter:
record.msg = self.formatter.format(record)
def do_emit(self, record):
# 具体输出逻辑，如写入文件或控制台
pass关键点：
- 处理器负责格式化和输出日志
setFormatter()方法绑定格式器emit()方法实现日志输出逻辑
七、进阶使用
7.1 日志轮转
import logging
from logging.handlers import RotatingFileHandler
# 配置日志轮转
handler = RotatingFileHandler('app.log', maxBytes=1024*1024, backupCount=5)
handler.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
# 记录大量日志
for i in range(1000):
logger.info(f"日志条目 {i}")关键点：
maxBytes控制文件大小backupCount控制备份文件数量- 自动轮转防止日志文件过大
7.2 异步日志处理
import logging
from logging.handlers import QueueHandler, QueueListener
# 创建队列
queue = Queue()
# 创建处理器
handler = logging.FileHandler('async.log')
handler.setLevel(logging.INFO)
# 创建队列处理器
queue_handler = QueueHandler(queue)
# 创建监听器
listener = QueueListener(queue, handler)
# 启动监听器
listener.start()
# 创建日志记录器
logger = logging.getLogger(__name__)
logger.addHandler(queue_handler)
logger.setLevel(logging.INFO)
# 异步记录日志
for i in range(100):
logger.info(f"异步日志 {i}")关键点：
- 使用
QueueHandler和QueueListener实现异步处理 - 避免阻塞主线程
- 适用于高性能要求场景
八、性能与工程实践
8.1 性能优化
| 优化策略 | 说明 | 适用场景 |
|---|---|---|
| 日志级别控制 | 通过设置日志级别过滤无关日志 | 生产环境 |
| 异步处理 | 使用队列机制避免阻塞 | 高并发系统 |
| 日志轮转 | 防止日志文件过大 | 长期运行系统 |
| 压缩归档 | 旧日志文件压缩存储 | 存储空间有限场景 |
8.2 安全风险
- 敏感信息泄露：日志中可能包含密码、API密钥等敏感信息
- 日志文件暴露：未授权访问日志文件可能导致信息泄露
- 日志注入攻击：用户输入未过滤可能导致日志文件被篡改
解决方案：
- 使用
%(message)s格式化字段避免任意字符串插入 - 设置合适的文件权限
- 使用
Filter过滤敏感信息
8.3 线程安全
import logging
import threading
# 创建日志记录器
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
# 创建文件处理器
handler = logging.FileHandler('thread_safe.log')
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(threadName)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
# 多线程日志记录
def worker():
for i in range(5):
logger.info(f"线程 {threading.current_thread().name} - 日志 {i}")
# 启动多个线程
threads = []
for i in range(4):
t = threading.Thread(target=worker, name=f"Thread-{i}")
t.start()
threads.append(t)
# 等待线程完成
for t in threads:
t.join()关键点：
logging模块是线程安全的- 使用
%(threadName)s记录线程信息 - 避免在多线程环境中使用
print()等非线程安全方法
九、常见问题与踩坑
9.1 日志不输出
可能原因：
- 日志级别设置错误（如设置为
ERROR而记录的是DEBUG） - 处理器未正确绑定
- 文件权限问题导致无法写入
解决方案：
# 检查日志级别
logger.setLevel(logging.DEBUG)
# 检查处理器
print(logger.handlers)
# 检查文件权限
os.chmod('app.log', 0o666)9.2 日志格式异常
错误示例：
formatter = logging.Formatter('%(asctime)s - %(message)s') # 缺少日志级别改进方案：
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')9.3 日志文件过大
解决方案：
- 使用
RotatingFileHandler自动轮转 - 设置
maxBytes和backupCount参数 - 定期清理旧日志文件
9.4 异步日志未生效
常见错误：
# 忘记启动监听器
listener.start()正确做法：
# 创建监听器并启动
listener = QueueListener(queue, handler)
listener.start()十、最佳实践
| 场景 | 推荐做法 | 原因 |
|---|---|---|
| 生产环境 | 设置ERROR级别 | 减少日志量，提高性能 |
| 开发调试 | 使用DEBUG级别 | 获取详细信息 |
| 跨模块日志 | 使用命名空间 | 方便分类管理 |
| 敏感信息 | 使用过滤器 | 避免泄露 |
| 高并发系统 | 使用异步处理 | 避免阻塞 |
十一、总结
Python的logging模块是一个功能强大但复杂的日志系统，其核心在于层次化设计和灵活配置。通过理解日志记录器、处理器、格式器之间的协作关系，可以构建出符合业务需求的日志系统。
在实际项目中，应根据场景选择适当的日志级别和处理器类型，注意日志安全和性能优化。对于复杂的日志需求，建议使用配置文件进行管理，避免在代码中硬编码配置。同时，要特别注意日志格式的规范性，防止因格式错误导致日志信息丢失。
掌握logging模块的高级特性，如日志轮转、异步处理、多线程支持等，可以显著提升系统的可观测性和可维护性。通过合理的设计和实践，日志系统将成为软件开发中不可或缺的利器。