1. 基础概念 1.1 logging 模块的作用 logging 模块用于为应用程序提供灵活的日志记录功能；
特点：通过threading locks 实现thread-safe，提到了threading lock 不全是可重入的*（TODO锁）*
1.2 logging 的基本组件
Logger : 负责记录日志消息，通常在应用程序的不同部分使用不同的 Logger 对象来进行日志记录。Handler : 负责将日志消息输出到指定的目标。常见的 Handler 包括 StreamHandler、FileHandler 等。
FileHandler 将消息发送到磁盘文件StreamHandler 消息发送到流RotatingFileHandler 发送到磁盘文件，支持最大日志文件大小和日志文件旋转
Formatter : 负责格式化日志消息，定义了日志的输出格式
时间%(asctime)s、函数名%(funcName)s、日志信息%(message)s等
Filter : 用于在日志处理过程中筛选特定的日志记录。它可以限制哪些日志消息应该被处理或者输出。
Logger 以Formatter的形式生成一条RecordFilter 在消息被输出之前筛选。Handler 通过不同的输出方式处理这些日志消息。
1.3 logging 的五个日志级别 Python logging 模块提供了五个标准的日志级别，它们用于表示日志消息的重要性和紧急性：
DEBUG : 用于输出详细的调试信息，通常仅在开发过程中使用。INFO : 用于输出常规的运行信息，标志程序的正常执行。WARNING : 用于输出警告信息，标志某些不影响程序运行的问题。ERROR : 用于输出错误信息，通常表示某些功能失败。CRITICAL : 用于输出严重错误信息，通常表示程序无法继续运行。
每个日志级别都可以输出该级别及更高级别的日志。例如，设置为 WARNING 级别时，将同时输出 WARNING、ERROR 和 CRITICAL 级别的日志。
2. 基本用法 2.1 Logger Logger 是日志系统的核心组件，用于记录日志消息。每个 Logger 都有一个名称，并且可以通过层级关系继承父级 Logger 的配置。Logger 的主要功能包括：
设置日志级别（如 DEBUG、INFO、WARNING 等）。
添加一个或多个 Handler，用于将日志消息输出到不同的目标。
可以通过 extra 参数传递自定义字段。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 import logginglogger = logging.getLogger("example" ) logger.setLevel(logging.DEBUG) handler = logging.StreamHandler() logger.addHandler(handler) logger.debug("This is a debug message" ) logger.info("This is an info message" ) logger.warning("This is a warning message" ) logger.error("This is an error message" ) logger.critical("This is a critical message" )
2.2 Handler 1 2 3 4 5 6 7 from logging.handlers import RotatingFileHandler # 创建一个 RotatingFileHandler file_handler = RotatingFileHandler( "example.log", maxBytes=1024 * 1024 * 5, backupCount=5 ) logger.addHandler(file_handler)
1 2 3 4 5 6 7 import logging formatter = logging.Formatter( fmt="%(asctime)s - %(levelname)s - %(name)s - %(funcName)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) handler.setFormatter(formatter)
自定义字段进行传参
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 import logging # 创建一个 Formatter，包含自定义字段 formatter = logging.Formatter( fmt="%(asctime)s - %(levelname)s - %(name)s - %(funcName)s - %(custom_field)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) # 创建一个 Logger logger = logging.getLogger("example") logger.setLevel(logging.DEBUG) # 定义一个函数，用于生成日志 def log_something(): # 使用 extra 参数传递自定义字段 extra_data = {"custom_field": "This is a custom field"} logger.info("This is an info message", extra=extra_data) log_something()
2.4 Filter 1 2 3 4 5 6 7 8 9 10 11 12 import logging class LevelFilter(logging.Filter): def __init__(self, level): super().__init__() self.level = level def filter(self, record): return record.levelno == self.level info_filter = LevelFilter(logging.INFO) handler.addFilter(info_filter)
3. 实践场景：gunicorn部署的Flask应用使用logging 需求：
gunicorn部署，多进程写入同一个文件而非创建不同文件
日志自动翻转，翻转后的日志采取日期命名，支持滚动删除
日志记录funcName与request_id
多进程日志库
1 pip install ConcurrentLogHandler==0.9.1
3.1 重写handler 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 import os import re import time from concurrent_log_handler import ConcurrentRotatingFileHandler from typing import List class CustomConcurrentRotatingFileHandler(ConcurrentRotatingFileHandler): """ 自定义日志处理器，继承 ConcurrentRotatingFileHandler， 增加自动命名日志文件及清理旧日志文件的功能。 """ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 直接继承父类的所有参数 self.dir_folder = os.path.dirname(self.baseFilename) self.default_filename = os.path.splitext(os.path.basename(self.baseFilename))[0] self.back_up_file_format = f"{self.default_filename}-*.log" def doRollover(self): """ 执行日志轮转，将当前日志文件重命名为带有时间戳的新文件， 并删除超出 backupCount 限制的最旧备份文件。 """ if self.stream: self.stream.close() self.stream = None # 生成新的备份日志文件名：/folder/OnlineServer-YYYYMMDD_HHMMSS.log current_time = time.strftime("%Y%m%d_%H%M%S") dfn = self.rotation_filename(os.path.join(self.dir_folder, f"{self.default_filename}-{current_time}.log")) if os.path.exists(dfn): os.remove(dfn) self.rotate(self.baseFilename, dfn) if not self.delay: self.stream = self._open() # 清理旧日志文件 if self.backupCount > 0: for file in self.getFilesToDelete(): os.remove(file) def getFilesToDelete(self) -> List[str]: """ 获取需要删除的旧日志文件 """ log_files = [f for f in os.listdir(self.dir_folder) if re.match(self.back_up_file_format.replace('*', '.*'), f)] log_files = [os.path.join(self.dir_folder, f) for f in log_files] if len(log_files) <= self.backupCount: return [] log_files.sort() return log_files[: len(log_files) - self.backupCount]
3.2 配置多进程logger 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 import logging import os from flask import g, has_request_context from kernel.logging.handler import CustomConcurrentRotatingFileHandler from config.consts import LOG_BASE_DIR, LOG_BASE_NAME, LOG_FILE_SIZE, LOG_BACK_UP_NUM, LOG_MYSQL_NAME, LOG_REDIS_NAME, \ LOG_NOS_NAME, LOG_SPLIT_NAME # 文件单例，多进程共用一个logger os.makedirs(LOG_BASE_DIR, exist_ok=True) logger = logging.getLogger('logger') # 自定义 Filter 来添加 request_id, 用于追踪与request_id相关的日志条目 class RequestIDFilter(logging.Filter): def filter(self, record): if has_request_context(): # 在请求上下文中安全获取 request_id record.request_id = getattr(g, 'request_id', 'no-request-id') else: # 如果没有请求上下文，尝试从日志记录的 extra 参数中获取 request_id # 假设在记录日志时通过 extra={'request_id': 'some-id'} 传递了 request_id record.request_id = getattr(record, 'request_id', 'non-request-id') return True def setup_loggers(): # logger使用自定义日志处理器，将日志写入 .log 文件中， # 当文件达到 LOG_FILE_SIZE MB 时进行轮转，最多保留 K 个备份 log_handler = CustomConcurrentRotatingFileHandler( os.path.join(LOG_BASE_DIR, LOG_BASE_NAME), "a", LOG_FILE_SIZE * 1024 * 1024, LOG_BACK_UP_NUM, encoding='utf-8') log_formatter = logging.Formatter( f'%(asctime)s - %(name)s - %(levelname)s - %(request_id)s - %(funcName)s - %(message)s') log_handler.setFormatter(log_formatter) logger.addHandler(log_handler) logger.setLevel(logging.INFO) logger.addFilter(RequestIDFilter()) logger.propagate = False setup_loggers()
参考：
logging — Logging facility for Python — Python 3.13.2 documentation
https://docs.python.org/3/howto/logging.html#logging-advanced-tutorial