《Python实战进阶》No27: 日志管理：Logging 模块的最佳实践（下)

No27: 日志管理：Logging 模块的最佳实践（下）

---

实战案例 ：复杂场景下的 Logging 配置与使用

本实战案例在 Python 3.11.5环境下运行通过

在本案例中，我们将通过一个复杂的日志配置示例，全面展示 logging 模块的参数设置和高级功能。该示例将涵盖以下内容：

• 使用配置文件（JSON 格式）动态调整日志设置。
• 同时输出日志到控制台、文件和远程服务器。
• 设置日志轮转策略。
• 自定义日志格式。
  

---

代码实现

1. 配置文件 (logging_config.json)

我们首先创建一个 JSON 文件来定义日志的详细配置。

{"version": 1,"disable_existing_loggers": false,"formatters": {"detailed": {"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"},"simple": {"format": "%(levelname)s - %(message)s"}},"handlers": {"console": {"class": "logging.StreamHandler","level": "DEBUG","formatter": "simple","stream": "ext://sys.stdout"},"file": {"class": "logging.handlers.RotatingFileHandler","level": "INFO","formatter": "detailed","filename": "app.log","maxBytes": 1048576,"backupCount": 3},"remote": {"class": "logging.handlers.SocketHandler","level": "ERROR","host": "localhost","port": 9020}},"loggers": {"app_logger": {"level": "DEBUG","handlers": ["console", "file", "remote"],"propagate": false}},"root": {"level": "WARNING","handlers": ["console"]}
}

2. Python 脚本 (complex_logging.py)

接下来，我们编写 Python 脚本来加载配置文件并使用日志系统。

import logging
import logging.config
import json
import os
import sys# 加载日志配置文件
def setup_logging(config_file):"""从 JSON 配置文件加载日志设置。"""try:if not os.path.exists(config_file):raise FileNotFoundError(f"配置文件 {config_file} 未找到。")with open(config_file, 'r', encoding='utf-8') as f:config = json.load(f)# 应用配置logging.config.dictConfig(config)print(f"成功加载日志配置: {config_file}")except json.JSONDecodeError as e:sys.stderr.write(f"配置文件格式错误: {e}\n")sys.exit(1)except Exception as e:sys.stderr.write(f"设置日志时出错: {e}\n")sys.exit(1)# 使用更安全的方式初始化日志
def initialize_logging():try:setup_logging('logging_config.json')return logging.getLogger('app_logger')except Exception as e:# 如果配置失败，设置基本日志print(f"使用默认日志配置: {e}")logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')return logging.getLogger('app_logger')# 初始化日志
logger = initialize_logging()# 示例函数：模拟程序运行
def simulate_application():try:logger.debug("这是一条调试信息。")       # 控制台和文件中可见logger.info("这是一条信息消息。")        # 控制台和文件中可见logger.warning("这是一条警告消息。")     # 控制台和文件中可见logger.error("这是一条错误消息。")       # 控制台、文件和远程服务器中可见logger.critical("这是一条严重错误消息。") # 控制台、文件和远程服务器中可见# 模拟一个异常情况try:result = 10 / 0except Exception as e:logger.exception(f"发生异常: {e}")except Exception as e:print(f"记录日志时出错: {e}")if __name__ == '__main__':simulate_application()print("日志测试完成。请检查日志文件和远程服务器。")# 为了确保所有日志都被发送，特别是通过网络logging.shutdown()

---

代码详解

1. 配置文件解析

• formatters: 定义了两种日志格式：
  • detailed: 包含时间戳、日志名称、日志级别和消息。
  • simple: 仅包含日志级别和消息。
• handlers:
  • console: 将日志输出到控制台，使用 simple 格式。
  • file: 将日志写入文件，支持日志轮转（最大 1MB，保留 3 个备份），使用 detailed 格式。
  • remote: 将错误日志发送到远程服务器（假设运行了一个监听端口 9020 的服务）。
• loggers:
  • app_logger: 自定义日志记录器，绑定到 console、file 和 remote 处理器。
  • root: 全局日志记录器，默认仅输出警告及以上级别的日志到控制台。
• disable_existing_loggers: 禁用默认的日志记录器，避免干扰。

2. 动态加载配置

通过 logging.config.dictConfig() 方法动态加载 JSON 配置文件，使日志配置更加灵活且易于维护。

3. 日志轮转

RotatingFileHandler 用于限制日志文件大小，并自动分割旧日志文件。例如，当 app.log 达到 1MB 时，会生成 app.log.1、app.log.2 等备份文件。

4. 远程日志

SocketHandler 将错误日志发送到远程服务器，适合分布式系统中的集中日志管理。

---

输入与输出

运行环境准备

• 启动远程日志服务器:
  假设我们有一个运行在 localhost:9020 的简单 TCP 日志服务器。

import socketserver
import pickle
import logging
import structclass LogRecordStreamHandler(socketserver.StreamRequestHandler):"""处理由logging.handlers.SocketHandler发送的日志记录"""def handle(self):"""处理来自客户端的日志记录"""while True:try:# 根据logging.handlers.SocketHandler的协议读取数据长度chunk = self.connection.recv(4)if len(chunk) < 4:breakslen = struct.unpack('>L', chunk)[0]chunk = self.connection.recv(slen)while len(chunk) < slen:chunk = chunk + self.connection.recv(slen - len(chunk))# 安全地反序列化数据obj = self.unPickle(chunk)record = logging.makeLogRecord(obj)# 打印日志信息print(f"收到日志: [{record.levelname}] {record.name}: {record.getMessage()}")# 可选：重新格式化并记录到本地日志系统# self.handleLogRecord(record)except Exception as e:print(f"处理日志时出错: {e}")breakdef unPickle(self, data):"""以更安全的方式反序列化数据"""try:return pickle.loads(data)except Exception as e:print(f"反序列化出错: {e}")raiseclass LogRecordSocketReceiver(socketserver.ThreadingTCPServer):"""使用TCP协议接收日志记录的简单服务器"""allow_reuse_address = Truedef __init__(self, host='localhost', port=9020, handler=LogRecordStreamHandler):socketserver.ThreadingTCPServer.__init__(self, (host, port), handler)self.abort = 0self.timeout = 1if __name__ == "__main__":# 启动日志服务器tcpserver = LogRecordSocketReceiver()print("日志服务器已在 localhost:9020 启动")try:tcpserver.serve_forever()except KeyboardInterrupt:print("服务器关闭中...")finally:tcpserver.server_close()

运行代码

执行主程序：

python complex_logging.py

输出结果

• 控制台输出:

成功加载日志配置: logging_config.json
DEBUG - 这是一条调试信息。
INFO - 这是一条信息消息。
WARNING - 这是一条警告消息。
ERROR - 这是一条错误消息。
CRITICAL - 这是一条严重错误消息。
ERROR - 发生异常: division by zero
Traceback (most recent call last):File "D:\python_projects\python_shizhan\logging_demo\complex_logging.py", line 57, in simulate_applicationresult = 10 / 0~~~^~~
ZeroDivisionError: division by zero
日志测试完成。请检查日志文件和远程服务器。

• 日志文件 (app.log):

2025-03-19 17:33:44,890 - app_logger - INFO - 这是一条信息消息。
2025-03-19 17:33:44,890 - app_logger - WARNING - 这是一条警告消息。
2025-03-19 17:33:44,890 - app_logger - ERROR - 这是一条错误消息。
2025-03-19 17:33:45,903 - app_logger - CRITICAL - 这是一条严重错误消息。
2025-03-19 17:33:45,904 - app_logger - ERROR - 发生异常: division by zero
Traceback (most recent call last):File "D:\python_projects\python_shizhan\logging_demo\complex_logging.py", line 57, in simulate_applicationresult = 10 / 0~~~^~~
ZeroDivisionError: division by zero

• 远程日志服务器:

日志服务器已在 localhost:9020 启动
收到日志: [ERROR] app_logger: 这是一条错误消息。
收到日志: [CRITICAL] app_logger: 这是一条严重错误消息。
收到日志: [ERROR] app_logger: 发生异常: division by zero

---

总结

本案例展示了如何通过 JSON 配置文件灵活地管理日志系统，并结合多种处理器实现多目标输出（控制台、文件、远程服务器）。同时，我们还介绍了日志轮转和远程日志发送等高级功能。这种配置方式非常适合复杂的生产环境，能够显著提升日志管理的效率和可维护性。

---

扩展思考

1. 如何优化日志性能？

• 在高并发场景下，可以使用异步日志处理器（如 QueueHandler 和 QueueListener）以减少日志记录对主线程的影响。
• 对于大文件日志，建议定期归档或压缩旧日志。

2. 如何增强日志安全性？

• 在发送日志到远程服务器时，建议启用加密传输（如 TLS/SSL）。
• 敏感信息（如用户密码）应从日志中过滤掉，避免泄露。

希望本集内容能帮助你更全面地掌握 logging 模块的高级用法！