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 / 0
except 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 struct
class LogRecordStreamHandler(socketserver.StreamRequestHandler):
"""处理由logging.handlers.SocketHandler发送的日志记录"""
def handle(self):
"""
处理来自客户端的日志记录
"""
while True:
try:
# 根据logging.handlers.SocketHandler的协议读取数据长度
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = 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}")
break
def unPickle(self, data):
"""
以更安全的方式反序列化数据
"""
try:
return pickle.loads(data)
except Exception as e:
print(f"反序列化出错: {e}")
raise
class LogRecordSocketReceiver(socketserver.ThreadingTCPServer):
"""
使用TCP协议接收日志记录的简单服务器
"""
allow_reuse_address = True
def __init__(self, host='localhost', port=9020, handler=LogRecordStreamHandler):
socketserver.ThreadingTCPServer.__init__(self, (host, port), handler)
self.abort = 0
self.timeout = 1
if __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_application
result = 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_application
result = 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 模块的高级用法！
