logging日志系统深入
构建可靠的Python日志系统
一、概述
Python内置的logging模块是一个功能完善、高度可扩展的日志系统,广泛应用于各种规模的Python项目中。与简单的print()调试不同,logging模块提供了等级控制、输出目标管理、格式自定义、过滤机制等企业级特性,是构建可靠Python应用的基石设施。
为什么选择logging而非print?
- 等级控制:可精细控制哪些信息需要输出(调试/信息/警告/错误),无需注释/取消注释print语句
- 多目标输出:同一日志可同时输出到控制台、文件、网络Socket、电子邮件等
- 格式统一:全局统一的时间格式、调用位置、线程信息等
- 性能优化:等级低于阈值的日志几乎零开销(通过等级判断过滤,不执行字符串格式化)
- 可配置性:支持通过代码、配置文件、JSON字典三种方式动态调整
logging模块的设计遵循组件化架构,核心由Logger(记录器)、Handler(处理器)、Formatter(格式化器)、Filter(过滤器)四大组件构成。理解这四大组件各自的职责与协作方式,是掌握logging模块的关键。
快速上手:最简单的日志配置
使用logging.basicConfig()一行即可完成基本配置,适合小型脚本和快速原型开发。
二、日志等级体系
logging模块定义了五个标准日志等级,从低到高依次为:DEBUG、INFO、WARNING、ERROR、CRITICAL。每个等级代表不同的严重程度,开发者可以根据需要选择在何种情况下触发日志记录。
| 等级 |
数值 |
适用场景 |
示例 |
| DEBUG |
10 |
调试信息,诊断问题时使用 |
变量值、函数入口/出口、SQL语句 |
| INFO |
20 |
程序正常运行的信息 |
启动完成、请求处理、定时任务触发 |
| WARNING |
30 |
表明潜在问题,但程序仍能运行 |
磁盘空间不足、已弃用API调用、配置缺失使用默认值 |
| ERROR |
40 |
由于严重问题,程序无法执行某些功能 |
数据库连接失败、文件写入异常、第三方服务超时 |
| CRITICAL |
50 |
严重错误,程序可能无法继续运行 |
内存耗尽、关键依赖缺失、数据库崩溃 |
等级过滤机制:Logger和Handler各自维护一个等级阈值。当日志记录的等级低于阈值时,该日志将被丢弃,不会进行后续处理。这种分层过滤机制可以非常灵活地控制日志输出。例如,可以将Logger设为DEBUG等级(记录所有日志),但将FileHandler设为WARNING等级(只将WARNING及以上写入文件),同时将StreamHandler设为INFO等级(控制台显示INFO及以上)。
import logging
# 自定义日志等级(高级用法)
CUSTOM_LEVEL = 25
logging.addLevelName(CUSTOM_LEVEL, "NOTICE")
def notice(self, message, *args, **kwargs):
if self.isEnabledFor(CUSTOM_LEVEL):
self._log(CUSTOM_LEVEL, message, args, **kwargs)
logging.Logger.notice = notice
# 验证等级数值关系
assert logging.DEBUG < logging.INFO < logging.WARNING < logging.ERROR < logging.CRITICAL
assert logging.getLevelName(10) == "DEBUG"
assert logging.getLevelName("WARNING") == 30
注意:等级数值空间
虽然Python允许自定义等级(如上面的NOTICE=25),但建议慎重使用。自定义等级破坏了标准的等级语义,可能导致团队协作中的理解混乱。大多数场景下,五个标准等级已足够。
三、四大组件详解
3.1 Logger(记录器)
Logger是应用程序直接使用的日志入口,负责产生日志记录。每个Logger实例都有一个名称,名称使用点号分隔的层级结构(如app.module.submodule),形成Logger的继承树。
import logging
# 获取Logger实例的推荐方式
logger = logging.getLogger(__name__) # 使用模块名自动命名
root_logger = logging.getLogger() # 获取根Logger(无参数)
app_logger = logging.getLogger("app") # 命名Logger
child_logger = logging.getLogger("app.module") # 子Logger
# Logger核心方法
logger.debug("这是调试信息: %s", value)
logger.info("用户 %s 登录成功", username)
logger.warning("磁盘剩余空间不足: %.1f GB", free_space)
logger.error("数据库连接失败: %s", exc_info=True)
logger.critical("系统内存耗尽,准备关闭")
logger.log(logging.INFO, "动态等级日志") # 动态指定等级
# 检查是否启用某等级(性能优化)
if logger.isEnabledFor(logging.DEBUG):
logger.debug("昂贵的计算: %s", expensive_function())
Logger命名最佳实践
- 在模块级别使用
logger = logging.getLogger(__name__),自动获得package.module形式的名称
- 不要直接实例化Logger类,始终通过
logging.getLogger()获取(工厂函数保证单例)
- 同一名称多次调用getLogger()返回同一个Logger实例
3.2 Handler(处理器)
Handler负责将日志记录发送到指定的目的地。一个Logger可以绑定多个Handler,将日志同时输出到不同目标。例如,开发时将日志输出到控制台和文件,生产环境额外发送到日志收集服务。
import logging
logger = logging.getLogger("app")
logger.setLevel(logging.DEBUG)
# Handler 1: 控制台输出
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(logging.Formatter("%(message)s"))
logger.addHandler(console_handler)
# Handler 2: 文件输出
file_handler = logging.FileHandler("app.log", encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"
))
logger.addHandler(file_handler)
# 避免重复日志:检查是否已有处理器
if not logger.handlers:
logger.addHandler(console_handler)
logger.addHandler(file_handler)
常见陷阱:重复日志
在模块级别创建Logger并添加Handler时,如果模块被多次导入(或使用basicConfig被重复调用),会导致日志重复输出。解决方法有二:①在if not logger.handlers检查后添加;②使用logging.lastResort机制或dictConfig集中配置。
3.3 Formatter(格式化器)
Formatter定义了日志记录的输出格式,使用%(attribute)s占位符语法。Python提供了丰富的内置属性,也支持自定义属性注入。
| 属性 |
格式 |
说明 |
| asctime |
%(asctime)s |
日志时间(可自定义datefmt) |
| levelname |
%(levelname)-8s |
日志等级(左对齐,宽度8) |
| name |
%(name)s |
Logger名称 |
| message |
%(message)s |
日志消息正文 |
| pathname |
%(pathname)s |
调用日志记录的源文件完整路径 |
| filename |
%(filename)s |
调用日志记录的文件名(不含路径) |
| funcName |
%(funcName)s |
调用日志记录的函数名 |
| lineno |
%(lineno)d |
调用日志记录的行号 |
| threadName |
%(threadName)s |
线程名称 |
| process |
%(process)d |
进程ID |
# 标准格式
fmt_standard = logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
# 简洁格式(生产环境控制台推荐)
fmt_compact = logging.Formatter(
"%(levelname)-8s %(message)s"
)
# 详细格式(文件日志推荐)
fmt_verbose = logging.Formatter(
"%(asctime)s.%(msecs)03d | %(levelname)-8s | %(threadName)-12s | "
"%(name)s:%(funcName)s:%(lineno)d | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
# 自定义Formatter:注入额外上下文
class ContextFormatter(logging.Formatter):
def __init__(self, fmt=None, datefmt=None, *, defaults={}):
super().__init__(fmt, datefmt)
self.defaults = defaults
def format(self, record):
record.__dict__.update(self.defaults)
return super().format(record)
# 使用:所有日志自动附加环境标识
context_fmt = ContextFormatter(
"%(asctime)s [%(env)s] %(message)s",
defaults={"env": "production"}
)
3.4 Filter(过滤器)
Filter提供了比等级更细粒度的日志过滤能力。它允许基于日志记录的任意属性(如Logger名称、消息内容、自定义上下文)来决定是否输出某条日志。
import logging
import random
# Filter 1: 按Logger名称过滤
class ModuleFilter(logging.Filter):
def __init__(self, allowed_module):
super().__init__()
self.allowed_module = allowed_module
def filter(self, record):
return record.name.startswith(self.allowed_module)
# Filter 2: 日志采样(仅输出30%的DEBUG日志)
class SamplingFilter(logging.Filter):
def __init__(self, sample_rate=0.3):
super().__init__()
self.sample_rate = sample_rate
def filter(self, record):
if record.levelno == logging.DEBUG:
return random.random() < self.sample_rate
return True
# Filter 3: 敏感信息脱敏
class SensitiveDataFilter(logging.Filter):
def __init__(self, patterns):
super().__init__()
self.patterns = patterns
def filter(self, record):
msg = record.getMessage()
for pattern, replacement in self.patterns.items():
msg = msg.replace(pattern, replacement)
record.msg = msg
record.args = ()
return True
# 使用过滤器
handler = logging.StreamHandler()
handler.addFilter(SamplingFilter(sample_rate=0.1))
handler.addFilter(SensitiveDataFilter({"password=123456": "password=***"}))
四、Handler类型详解
logging模块内置了十余种Handler,覆盖了从文件输出到网络传输的各种场景。选择合适的Handler类型是构建日志系统的核心环节。
4.1 StreamHandler(流处理器)
最基础的Handler,将日志输出到任意流对象(默认sys.stderr)。所有其他Handler本质上都是StreamHandler的扩展。
import sys
import logging
# 输出到标准错误(默认)
h1 = logging.StreamHandler()
# 输出到标准输出
h2 = logging.StreamHandler(sys.stdout)
# 输出到字符串缓冲区
from io import StringIO
buf = StringIO()
h3 = logging.StreamHandler(buf)
4.2 FileHandler(文件处理器)
将日志写入单个文件,支持指定编码和写入模式。
# 基础文件日志
fh = logging.FileHandler("app.log", mode="a", encoding="utf-8")
# 按天分文件(手动管理方式)
from datetime import datetime
class DailyFileHandler(logging.FileHandler):
def __init__(self, dirname, basename, encoding="utf-8"):
self.dirname = dirname
self.basename = basename
filename = self._get_filename()
super().__init__(filename, encoding=encoding)
def _get_filename(self):
today = datetime.now().strftime("%Y-%m-%d")
return f"{self.dirname}/{self.basename}_{today}.log"
4.3 RotatingFileHandler(轮转文件处理器)
当日志文件达到指定大小(如10MB)时,自动轮转生成新文件,并保留指定数量的历史文件。这是生产环境最常用的文件Handler。
from logging.handlers import RotatingFileHandler
# 配置文件轮转:每个文件10MB,保留5个备份
rfh = RotatingFileHandler(
filename="app.log",
mode="a",
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5,
encoding="utf-8"
)
# 轮转效果:
# app.log → 当前日志
# app.log.1 → 最近一次轮转
# app.log.2 → 前一次轮转
# ... → 最多到 app.log.5
# 手动触发轮转
rfh.doRollover()
4.4 TimedRotatingFileHandler(定时轮转文件处理器)
按时间间隔轮转日志文件,支持按秒、分、时、天、周、月轮转。
from logging.handlers import TimedRotatingFileHandler
# 每天午夜轮转,保留30天历史
trfh = TimedRotatingFileHandler(
filename="app.log",
when="midnight", # 每天午夜轮转
interval=1,
backupCount=30,
encoding="utf-8"
)
# when参数选项:
# 'S' → 秒 (每interval秒轮转)
# 'M' → 分
# 'H' → 时
# 'D' → 天
# 'W0-W6'→ 周 (W0=周一, W6=周日)
# 'midnight' → 每天午夜
# 每小时轮转,保留72份(3天)
hourly_handler = TimedRotatingFileHandler(
filename="hourly/app.log",
when="H",
interval=1,
backupCount=72,
encoding="utf-8"
)
4.5 其他重要Handler
内置Handler一览
- SocketHandler:通过网络Socket发送日志,适用于集中式日志收集
- DatagramHandler:基于UDP的SocketHandler
- SysLogHandler:发送日志到UNIX syslog服务
- SMTPHandler:通过邮件发送ERROR及以上日志(适用于告警场景)
- HTTPHandler:通过POST/GET请求将日志发送到Web服务
- QueueHandler:将日志放入队列,与QueueListener配合实现异步日志
- NullHandler:丢弃所有日志,通常用于库的默认配置
from logging.handlers import (
SMTPHandler, QueueHandler, QueueListener, HTTPHandler
)
from queue import Queue
# SMTP邮件告警(仅在ERROR时触发)
smtp_handler = SMTPHandler(
mailhost=("smtp.example.com", 587),
fromaddr="monitor@example.com",
toaddrs=["ops@example.com"],
subject="[App ERROR] 系统异常告警",
credentials=("user", "password"),
secure=()
)
smtp_handler.setLevel(logging.ERROR)
# 异步日志:通过队列解耦
log_queue: Queue = Queue(-1)
queue_handler = QueueHandler(log_queue)
# 在独立线程中消费队列
file_handler = logging.FileHandler("async.log")
listener = QueueListener(log_queue, file_handler)
listener.start() # 启动后台线程
# 程序退出时清理
import atexit
atexit.register(listener.stop)
性能建议:异步日志
在高并发场景下,QueueHandler + QueueListener 组合至关重要。日志I/O操作被转移到独立的后台线程,主线程仅将日志放入内存队列,从而避免日志I/O阻塞业务逻辑。实测可降低日志操作对主线程的影响从毫秒级降至微秒级。
五、Logger继承层次与Propagate机制
Logger名称使用点号分隔的命名空间(类似Python包的层级),形成父子继承关系。例如,app.service.user 是 app.service 的子Logger,而 app.service 又是 app 的子Logger。所有Logger的根节点是root Logger。
继承规则
- 等级继承:子Logger未设置等级时,沿继承链向上查找最近的父Logger等级
- Handler传播:默认情况下(
propagate=True),子Logger的日志会传递给父Logger的Handler处理
- 避免重复:如果父Logger和子Logger都绑定了Handler,且
propagate=True,日志会被所有Handler处理一次,造成重复
import logging
# 设置根Logger
root = logging.getLogger()
root.setLevel(logging.WARNING)
root_handler = logging.StreamHandler()
root.addHandler(root_handler)
# 创建层级Logger
parent = logging.getLogger("app")
child = logging.getLogger("app.service")
grandchild = logging.getLogger("app.service.user")
# 验证继承关系
assert parent.parent is root
assert child.parent is parent
assert grandchild.parent is child
# propagate机制演示
child.info("这条日志不会显示") # 子Logger未设等级,继承parent的->parent未设,继承root的WARNING,INFO被过滤
child.setLevel(logging.DEBUG)
child.info("这条会显示(INFO>=DEBUG,且传播到root的Handler)")
# 关闭propagate避免传播
child.propagate = False
child.info("这条不会显示(propagate=False,且child本身没有Handler)")
# 给子Logger添加自己的Handler
child_handler = logging.StreamHandler()
child.addHandler(child_handler)
child.info("这条会显示(子Logger有自己的Handler)")
# 典型应用场景:按模块控制日志
import logging.config
LOGGING_CONFIG = {
"version": 1,
"handlers": {
"console": {"class": "logging.StreamHandler"},
"file": {"class": "logging.FileHandler", "filename": "app.log"},
},
"loggers": {
"app": { # app及其子Logger默认使用console
"handlers": ["console"],
"level": "INFO",
},
"app.api": { # API模块额外写文件,等级更严格
"handlers": ["file"],
"level": "WARNING",
"propagate": False, # 不传播到app,避免重复
},
},
"root": { # 根Logger兜底
"handlers": ["console"],
"level": "WARNING",
},
}
logging.config.dictConfig(LOGGING_CONFIG)
Propagate常见问题:日志重复
这是logging模块最常见的坑。当子Logger和父Logger都添加了Handler,且子Logger的propagate=True时,一条日志会被子Logger的Handler处理一次,再传播给父Logger被父Logger的Handler再处理一次,造成重复输出。解决办法:在子Logger上设置propagate = False,或者只在父Logger上配置Handler。
六、日志配置方式
logging模块支持三种配置方式,分别适用于不同场景。理解每种方式的优劣,有助于在项目中做出合理选择。
6.1 代码配置(basicConfig)
适用于小型脚本和快速原型。一行代码完成基础配置,简单直观,但灵活性有限。
import logging
# 最简配置:一行搞定
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
filename="app.log", # 可选:输出到文件
filemode="a",
encoding="utf-8"
)
# 注意事项:
# 1. basicConfig是幂等的——只有第一次调用生效
# 2. 如果根Logger已有Handler,调用无效果
# 3. 只能配置根Logger,无法配置子Logger
# 检查是否已配置
if not logging.getLogger().handlers:
logging.basicConfig(level=logging.INFO, ...)
6.2 文件配置(fileConfig)
将日志配置外置到配置文件(支持INI和YAML格式),便于运维人员在不修改代码的前提下调整日志策略。
# logging.ini 配置文件
# [loggers]
# keys=root,app
#
# [handlers]
# keys=consoleHandler,fileHandler
#
# [formatters]
# keys=simpleFormatter,detailedFormatter
#
# [logger_root]
# level=WARNING
# handlers=consoleHandler
#
# [logger_app]
# level=DEBUG
# handlers=fileHandler
# qualname=app
# propagate=0
#
# [handler_consoleHandler]
# class=StreamHandler
# level=INFO
# formatter=simpleFormatter
# args=(sys.stderr,)
#
# [handler_fileHandler]
# class=handlers.RotatingFileHandler
# level=DEBUG
# formatter=detailedFormatter
# args=('app.log','a',10000000,5,'utf-8')
#
# [formatter_simpleFormatter]
# format=%(levelname)-8s %(message)s
#
# [formatter_detailedFormatter]
# format=%(asctime)s | %(levelname)-8s | %(name)s | %(message)s
# datefmt=%%Y-%%m-%%d %%H:%%M:%%S
# Python代码加载INI配置文件
logging.config.fileConfig("logging.ini", disable_existing_loggers=False)
6.3 dictConfig(字典配置,推荐)
最强大、最灵活的配置方式。使用Python字典(可直接写在代码中或从JSON/YAML文件加载)描述完整的日志配置,支持所有特性的细粒度控制。这是生产环境推荐的方式。
import logging.config
import json
LOGGING_CONFIG = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"standard": {
"format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S",
},
"json": {
"class": "pythonjsonlogger.jsonlogger.JsonFormatter",
"format": "%(asctime)s %(levelname)s %(name)s %(message)s",
},
},
"filters": {
"sampling": {
"()": "app.logging_filters.SamplingFilter",
"sample_rate": 0.1,
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "standard",
"stream": "ext://sys.stdout",
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"level": "DEBUG",
"formatter": "json",
"filename": "logs/app.log",
"maxBytes": 10485760,
"backupCount": 5,
"encoding": "utf-8",
"filters": ["sampling"],
},
"error_file": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "ERROR",
"formatter": "standard",
"filename": "logs/error.log",
"when": "midnight",
"backupCount": 30,
"encoding": "utf-8",
},
},
"loggers": {
"app": {
"handlers": ["console", "file"],
"level": "DEBUG",
"propagate": False,
},
"app.api": {
"handlers": ["error_file"],
"level": "ERROR",
"propagate": False,
},
},
"root": {
"handlers": ["console"],
"level": "WARNING",
},
}
# 加载配置
logging.config.dictConfig(LOGGING_CONFIG)
# 从JSON文件加载
with open("logging.json", "r", encoding="utf-8") as f:
config = json.load(f)
logging.config.dictConfig(config)
配置方式对比总结
| 方式 |
灵活性 |
可维护性 |
适用场景 |
| basicConfig |
低 |
低 |
脚本、快速原型 |
| fileConfig (INI) |
中 |
高 |
传统项目、运维友好 |
| dictConfig |
高 |
高 |
生产环境、大型项目 |
七、结构化日志(JSON格式)
传统文本日志对人类阅读友好,但对日志分析系统(如ELK Stack、Splunk、Loki)而言,解析效率低下且容易出错。结构化日志将日志以JSON格式输出,每条日志的字段清晰可辨,便于自动化解析和检索。
# 安装:pip install python-json-logger
from pythonjsonlogger import jsonlogger
import logging
# 自定义JSON Formatter
class CustomJsonFormatter(jsonlogger.JsonFormatter):
def add_fields(self, log_record, record, message_dict):
super().add_fields(log_record, record, message_dict)
log_record["app_name"] = "my_app"
log_record["environment"] = "production"
log_record["host"] = "server-01"
# 配置使用JSON格式
handler = logging.StreamHandler()
formatter = CustomJsonFormatter(
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
handler.setFormatter(formatter)
logger = logging.getLogger("app")
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# 输出示例(实际JSON单行输出)
# {"asctime": "2026-05-05T22:48:47", "levelname": "INFO",
# "name": "app", "message": "用户登录成功",
# "app_name": "my_app", "environment": "production",
# "host": "server-01"}
logger.info("用户登录成功", extra={"user_id": 1001, "ip": "192.168.1.1"})
# 纯Python实现JSON格式化(不依赖第三方库)
import json
import logging
from datetime import datetime
class JsonFormatter(logging.Formatter):
def format(self, record):
log_obj = {
"timestamp": datetime.fromtimestamp(record.created).isoformat(),
"level": record.levelname,
"logger": record.name,
"module": record.module,
"function": record.funcName,
"line": record.lineno,
"message": record.getMessage(),
}
if record.exc_info and record.exc_info[0]:
log_obj["exception"] = self.formatException(record.exc_info)
if hasattr(record, "extra_fields"):
log_obj.update(record.extra_fields)
return json.dumps(log_obj, ensure_ascii=False)
# 使用extra传递额外字段
class ExtraLogger:
def __init__(self, logger):
self._logger = logger
def info(self, msg, **extra):
self._logger.info(msg, extra={"extra_fields": extra})
def error(self, msg, **extra):
self._logger.error(msg, extra={"extra_fields": extra})
logger = ExtraLogger(logging.getLogger("app"))
logger.info("订单已创建", order_id=10086, amount=299.00, currency="CNY")
结构化日志的优势
- 机器可解析:直接导入ELK、Loki等日志系统,无需正则解析
- 字段可搜索:Kibana中可直接按
level:ERROR AND user_id:1001 过滤
- 上下文丰富:每个日志条目自带完整上下文(trace_id、user_id、请求耗时等)
- 动态字段:不同日志类型可携带不同的额外字段,不受固定格式约束
八、日志轮转与归档
生产环境中,日志文件如果不加管理会无限增长,最终耗尽磁盘空间。日志轮转(Log Rotation)解决了这一问题,它允许你按大小或时间自动拆分日志文件,并清理过期历史文件。
8.1 基于文件大小的轮转
from logging.handlers import RotatingFileHandler
# 适用于:日志量稳定的场景(如API请求日志)
rfh = RotatingFileHandler(
filename="logs/api.log",
maxBytes=100 * 1024 * 1024, # 100MB
backupCount=10, # 保留10个备份
encoding="utf-8",
)
# 轮转策略:
# - 当前日志写入 api.log
# - api.log 达到100MB → 重命名为 api.log.1,新建 api.log
# - 原 api.log.1 → api.log.2,依此类推
# - api.log.10 被删除(超过backupCount)
8.2 基于时间的轮转
from logging.handlers import TimedRotatingFileHandler
# 按天轮转(适用于:业务日志,每天一个文件)
daily_handler = TimedRotatingFileHandler(
filename="logs/business.log",
when="midnight",
interval=1,
backupCount=90, # 保留90天
encoding="utf-8",
utc=False,
)
# 按小时轮转(适用于:高流量系统,便于精细化排查)
hourly_handler = TimedRotatingFileHandler(
filename="logs/debug.log",
when="H",
interval=6, # 每6小时轮转一次
backupCount=28, # 保留7天的数据(28 * 6h)
encoding="utf-8",
)
# 周轮转(适用于:归档日志)
weekly_handler = TimedRotatingFileHandler(
filename="logs/archive.log",
when="W0", # 每周一轮转
backupCount=52, # 保留1年
)
8.3 混合轮转策略
生产环境推荐:双层轮转
同时配置基于大小和时间的轮转,互为补充。例子:
- DEBUG日志:每小时轮转,保留72份(3天),用于快速排查问题
- INFO日志:每日轮转,保留30份(1月),用于日常监控
- ERROR日志:每日轮转,保留365份(1年),用于审计和事后分析
- 访问日志:按大小轮转(每100MB),保留50份,用于流量分析
# 生产环境综合配置示例
LOGGING_CONFIG = {
"version": 1,
"formatters": {
"detailed": {"format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s"},
},
"handlers": {
"debug": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "DEBUG",
"formatter": "detailed",
"filename": "/var/log/app/debug.log",
"when": "H",
"interval": 6,
"backupCount": 28,
},
"error": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "ERROR",
"formatter": "detailed",
"filename": "/var/log/app/error.log",
"when": "midnight",
"backupCount": 365,
},
},
"loggers": {
"app": {
"handlers": ["debug", "error"],
"level": "DEBUG",
},
},
"root": {"handlers": ["debug"], "level": "INFO"},
}
九、最佳实践
9.1 分层日志策略
推荐的分层方案
- 开发环境:DEBUG等级输出到控制台,带详细时间和位置信息
- 测试环境:INFO等级输出到控制台和文件,开启SQL日志
- 预发布环境:WARNING等级输出到控制台,INFO等级写入日志文件
- 生产环境:INFO等级写入JSON格式日志到集中式系统,ERROR触发告警
9.2 性能优化
import logging
logger = logging.getLogger("app")
# 反模式:即使DEBUG被禁用,expensive_func()仍被执行
logger.debug("调试信息: %s", expensive_func())
# 正模式:先判断等级,避免不必要的计算
if logger.isEnabledFor(logging.DEBUG):
logger.debug("调试信息: %s", expensive_func())
# 或者使用 lazy %s 格式化(推荐,args只在需要时计算)
logger.debug("用户数据: %s", get_user_data(user_id))
# 反模式:使用f-string(即使不输出也会格式化字符串)
logger.debug(f"用户 {user_id} 数据: {get_user_data(user_id)}")
9.3 日志安全
绝不能记录到日志的信息
- 密码和密钥:数据库密码、API密钥、JWT令牌、CSRF令牌
- 个人身份信息(PII):身份证号、银行卡号、完整电话号码
- 敏感商业数据:数据库连接字符串、内部网络拓扑
- 会话令牌:Session ID、Access Token(可记录掩码版本)
# 安全的日志脱敏
import re
def sanitize_for_log(message: str) -> str:
"""将消息中的敏感信息替换为掩码"""
message = re.sub(r"(password|passwd|secret)=\S+", r"\1=***", message, flags=re.I)
message = re.sub(r"\b\d{17}[\dXx]\b", "ID_CARD_MASKED", message)
message = re.sub(r"\b1[3-9]\d{9}\b", "PHONE_MASKED", message)
return message
class SanitizingFilter(logging.Filter):
def filter(self, record):
record.msg = sanitize_for_log(record.msg)
return True
9.4 综合示例:完整的生产环境日志系统
import logging
import logging.config
import sys
from pathlib import Path
# 确保日志目录存在
Path("logs").mkdir(exist_ok=True)
def setup_logging(env: str = "development"):
"""
根据环境配置日志系统
Args:
env: 运行环境(development / staging / production)
"""
config = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"simple": {
"format": "%(levelname)-8s %(message)s",
},
"detailed": {
"format": "%(asctime)s.%(msecs)03d | %(levelname)-8s | "
"%(name)s:%(funcName)s:%(lineno)d | %(message)s",
"datefmt": "%Y-%m-%d %H:%M:%S",
},
"json": {
"format": "%(asctime)s %(levelname)s %(name)s %(message)s",
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "DEBUG",
"formatter": "simple",
"stream": "ext://sys.stdout",
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"level": "DEBUG",
"formatter": "detailed",
"filename": "logs/app.log",
"maxBytes": 20 * 1024 * 1024,
"backupCount": 5,
"encoding": "utf-8",
},
"errors": {
"class": "logging.handlers.TimedRotatingFileHandler",
"level": "ERROR",
"formatter": "detailed",
"filename": "logs/error.log",
"when": "midnight",
"backupCount": 30,
"encoding": "utf-8",
},
},
"loggers": {
"app": {
"handlers": ["console", "file", "errors"],
"level": "DEBUG",
"propagate": False,
},
},
"root": {
"handlers": ["console"],
"level": "WARNING",
},
}
# 根据环境调整配置
if env == "production":
config["handlers"]["console"]["level"] = "WARNING"
config["handlers"]["console"]["formatter"] = "json"
config["loggers"]["app"]["level"] = "INFO"
elif env == "development":
config["handlers"]["console"]["level"] = "DEBUG"
config["loggers"]["app"]["level"] = "DEBUG"
logging.config.dictConfig(config)
# 使用示例
setup_logging(env="development")
logger = logging.getLogger("app")
logger.info("应用启动成功")
十、核心要点总结
- 四大组件:Logger(入口)、Handler(输出目标)、Formatter(格式)、Filter(过滤),各司其职,组合灵活
- 五级日志:DEBUG/INFO/WARNING/ERROR/CRITICAL,等级过滤是性能优化的第一道防线
- Handler选择:开发用StreamHandler,基础文件用FileHandler,轮转用RotatingFileHandler/TimedRotatingFileHandler
- 配置方式:小脚本用basicConfig,生产环境强烈推荐dictConfig,支持从JSON/YAML加载
- 继承与传播:Logger形成父子继承树,propagate控制是否向父Logger传递,注意避免重复日志
- Filter威力:不仅可按名称过滤,还能实现日志采样、脱敏、上下文注入等高级功能
- 结构化日志:JSON格式日志是现代日志系统(ELK/Loki)的基石,
python-json-logger 是最常用的第三方扩展
- 异步日志:
QueueHandler + QueueListener 在高并发场景下将日志I/O移出主线程
- 安全第一:永远不在日志中记录密码、密钥、PII等敏感信息,使用Filter统一脱敏
- 环境适配:不同环境(开发/测试/预发布/生产)使用不同的日志等级、格式和输出目标
十一、进一步思考
Python logging模块的设计借鉴了Apache Log4j的架构思想,是Java日志生态在Python世界的成功移植。理解logging模块的组件化设计,不仅能写出更好的日志代码,更能体会到分层抽象在系统设计中的普适价值。
扩展方向
- 分布式追踪:结合
opentelemetry,在日志中注入trace_id/span_id,实现跨服务的日志关联
- 日志告警:使用SMTPHandler或自定义Handler,在ERROR/CRITICAL时触发钉钉/飞书/企业微信机器人通知
- 日志采集:将日志发送到Filebeat/Fluentd,再由Logstash处理进入Elasticsearch
- 动态调级:通过管理接口在运行时动态修改Logger的等级,无需重启服务
- 上下文日志:使用
logging.LoggerAdapter或MDC(Mapped Diagnostic Context)模式,自动附加请求级别上下文
- structlog库:如果内置logging的灵活性仍不足,可以考虑
structlog第三方库,它提供了更现代的日志API
推荐的日志学习路线
- 掌握基础:logging模块四大组件、日志等级、basicConfig(1天)
- Handler精通:RotatingFileHandler、TimedRotatingFileHandler、QueueHandler(2天)
- 配置管理:dictConfig + JSON/YAML加载,环境分离配置(1天)
- 高级主题:自定义Filter、结构化日志、异步日志(2天)
- 生产实践:ELK集成、分布式追踪、日志告警(3天)