彩色logger先来一个再说细节:

# -*- coding: utf-8 -*-

import logging
import colorlog  # 需要安装：pip install colorlog
import os

from SmartVoyage2.config import Config

project_root = os.path.dirname(os.path.abspath(__file__))
log_file_path = os.path.join(project_root, Config().log_file)


def setup_logging(log_file=log_file_path):
    os.makedirs(os.path.dirname(log_file), exist_ok=True)
    logger = logging.getLogger("SmartVoyage")
    logger.setLevel(logging.INFO)
    
    if not logger.handlers:
        # 文件处理器（无颜色，存文件）
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        file_handler.setLevel(logging.INFO)
        file_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        file_handler.setFormatter(file_formatter)
        logger.addHandler(file_handler)
        
        # 控制台处理器（有颜色）
        console_handler = colorlog.StreamHandler()
        console_handler.setLevel(logging.INFO)
        console_formatter = colorlog.ColoredFormatter(
            '%(log_color)s%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            log_colors={
                'DEBUG': 'cyan',
                'INFO': 'green',
                'WARNING': 'yellow',
                'ERROR': 'red',
                'CRITICAL': 'red,bg_white',
            }
        )
        console_handler.setFormatter(console_formatter)
        logger.addHandler(console_handler)
    
    return logger


logger = setup_logging()

if __name__ == '__main__':
    logger.info('测试日志功能')  # 绿色
    logger.error('测试错误')     # 红色

logging等级

try:
    1 / 0
except Exception as e:
    logger.exception("计算错误")

"""
ERROR:test:计算错误
Traceback (most recent call last):
  File "test.py", line 6, in <module>
    1 / 0
ZeroDivisionError: division by zero

没有堆栈信息的话:
ERROR:test:计算错误
"""

生产环境调试全靠堆栈信息，logger.exception() 是定位问题的第一道防线。

# 数据库操作
try:
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
except Exception as e:
    logger.exception(f"查询用户失败, user_id={user_id}")
    return None

# API 调用
try:
    response = requests.get(f"https://api.example.com/user/{user_id}")
except Exception as e:
    logger.exception(f"调用用户API失败, user_id={user_id}")
    raise

# 文件处理
try:
    with open(file_path, 'r') as f:
        data = f.read()
except Exception as e:
    logger.exception(f"读取文件失败, path={file_path}")

一、代码全景展示

import logging
from logging.handlers import RotatingFileHandler
import os

def get_logger(name, level, log_file):
    """获取配置好的logger，支持日志轮转"""
    logger = logging.getLogger(name)
    logger.setLevel(level)
    
    # 避免重复添加handler
    if logger.handlers:
        return logger
    
    # 确保日志目录存在
    log_dir = os.path.dirname(log_file)
    if log_dir and not os.path.exists(log_dir):
        os.makedirs(log_dir)
    
    # 文件处理器（10MB轮转，保留5个备份）
    file_handler = RotatingFileHandler(
        log_file, maxBytes=10*1024*1024, backupCount=5, encoding='utf-8'
    )
    file_handler.setLevel(level)
    
    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setLevel(level)
    
    # 格式化器
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
    )
    file_handler.setFormatter(formatter)
    console_handler.setFormatter(formatter)
    
    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    
    return logger

二、逐行深度解析

第1-3行：导入必要的模块

import logging                    # Python标准日志库
from logging.handlers import RotatingFileHandler  # 日志轮转处理器
import os                         # 操作系统接口，用于路径处理

生产环境考量：

• logging是Python官方日志库，成熟稳定

• RotatingFileHandler解决单一日志文件过大问题

• os模块确保跨平台路径兼容性

第5行：函数定义

def get_logger(name, level, log_file):

参数说明：

参数	类型	说明	示例
name	str	日志器名称，通常用__name__	"user_service"
level	int	日志级别	logging.INFO
log_file	str	日志文件路径	"logs/app.log"

第6行：文档字符串

"""获取配置好的logger，支持日志轮转"""

生产环境必须有清晰的文档说明，便于团队协作。

第7-8行：创建并设置日志器

"""
对象已存在就获取,不存在就创建新的,保证单对象
"""
logger = logging.getLogger(name)  # 获取或创建logger实例

logger.setLevel(level)             # 设置日志级别过滤器

关键知识点：

• getLogger()是工厂方法模式，相同name返回同一个实例（单例）

• setLevel()设置全局过滤级别，低于此级别的日志不会处理

• 子文件夹会自动继承父文件夹的LEVEL,自己有用自己的,自己没有用父文件的

第10-12行：避免重复添加Handler（核心防护）

"""
保障单对象➕单配置
这里的handlers是一个list,放着我们的 文件处理器 和 控制台处理器
"""
# 避免重复添加handler
if logger.handlers:      # 检查是否已有处理器
    return logger        # 直接返回现有实例

生产环境重要性：

# 错误示例：没有防护的代码
get_logger("app", logging.INFO, "app.log")
get_logger("app", logging.INFO, "app.log")
# 结果：同一个logger会有2个相同的handler，日志会重复输出两次！

# 正确示例：有防护的代码
logger1 = get_logger("app", logging.INFO, "app.log")
logger2 = get_logger("app", logging.INFO, "app.log")
print(logger1 is logger2)  # True，handler只添加一次

第14-18行：自动创建日志目录

log_dir = os.path.dirname(log_file)  # 提取目录路径
if log_dir and not os.path.exists(log_dir):  # 目录不存在时
    os.makedirs(log_dir)             # 递归创建所有缺失的目录

log_file = "logs/2024/01/app.log"
log_dir = os.path.dirname(log_file)  # 结果："logs/2024/01"

# 自动创建 logs/2024/01 目录结构
# 避免手动创建目录导致的 FileNotFoundError

第20-24行：文件处理器（日志轮转核心）

file_handler = RotatingFileHandler(
    log_file,                      # 日志文件路径
    maxBytes=10*1024*1024,        # 10MB后轮转
    backupCount=5,                # 保留5个备份
    encoding='utf-8'              # UTF-8编码，支持中文
)
file_handler.setLevel(level)      # 设置处理器级别

轮转机制演示：

"""
初始状态：
app.log (5MB)

写入5MB后（总10MB）：
app.log → 轮转 → app.log.1
新建 app.log

继续写入，达到10MB时：
app.log → app.log.1
app.log.1 → app.log.2
... 以此类推

最多保留5个备份：
app.log (当前)
app.log.1 (最新备份)
app.log.2
app.log.3
app.log.4
app.log.5 (最旧备份)

核心是：备份数 = backupCount，总文件数 = backupCount + 1
"""

第26-28行：控制台处理器

console_handler = logging.StreamHandler()  # 输出到标准输出
console_handler.setLevel(level)            # 设置级别

生产环境用途：

• 开发调试时实时查看日志

• 容器化部署（Docker/K8s）时通过stdout收集日志

• CI/CD流水线中查看构建日志

第30-34行：日志格式化器

formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
)
file_handler.setFormatter(formatter)   # 文件使用完整格式
console_handler.setFormatter(formatter) # 控制台也使用相同格式

格式占位符详解：

占位符	说明	示例输出
%(asctime)s	时间戳	2024-01-15 10:30:45,123
%(name)s	日志器名称	user_service
%(levelname)s	日志级别	INFO / ERROR
%(filename)s	源文件名	user_controller.py
%(lineno)d	行号	42
%(message)s	日志消息	用户登录成功

输出示例：

"""
2024-01-15 10:30:45,123 - user_service - INFO - user_controller.py:42 - 用户登录成功
"""

第36-37行：组装并返回

logger.addHandler(file_handler)    # 添加文件处理器
logger.addHandler(console_handler) # 添加控制台处理器
return logger                      # 返回配置好的实例

三、完整使用示例

示例1：基础使用

# 初始化日志器
logger = get_logger(
    name="my_app",
    level=logging.INFO,
    log_file="logs/my_app.log"
)

# 使用不同级别记录日志
logger.debug("这是调试信息")      # 不会输出（INFO级别高于DEBUG）
logger.info("应用启动成功")       # ✅ 会输出
logger.warning("配置文件未找到，使用默认配置")  # ✅ 会输出
logger.error("数据库连接失败")     # ✅ 会输出
logger.critical("系统崩溃")        # ✅ 会输出

# 异常信息记录
try:
    1 / 0
except Exception as e:
    logger.exception("计算错误")   # 自动记录异常堆栈

示例2：多模块使用

# database.py
import logging
from logger_config import get_logger

logger = get_logger("database", logging.INFO, "logs/app.log")

def connect_db():
    logger.info("连接数据库...")
    # 数据库连接代码

# user_service.py
from logger_config import get_logger

logger = get_logger("user_service", logging.INFO, "logs/app.log")

def create_user(username):
    logger.info(f"创建用户: {username}")
    # 用户创建逻辑

输出效果：

"""
2024-01-15 10:30:45,123 - database - INFO - database.py:5 - 连接数据库...
2024-01-15 10:30:46,456 - user_service - INFO - user_service.py:8 - 创建用户: admin
"""

示例3：Web框架集成（Flask）

from flask import Flask, request
import logging

app = Flask(__name__)
logger = get_logger("web_app", logging.INFO, "logs/web.log")

@app.before_request
def log_request():
    logger.info(f"{request.method} {request.path} - IP: {request.remote_addr}")

@app.route('/api/user/<int:uid>')
def get_user(uid):
    logger.info(f"查询用户: {uid}")
    return {"user_id": uid}

if __name__ == '__main__':
    app.run()

四、生产环境进阶优化

优化1：支持环境变量配置

import os

def get_logger_advanced(name):
    log_level = getattr(logging, os.getenv('LOG_LEVEL', 'INFO'))
    log_file = os.getenv('LOG_FILE', 'logs/app.log')
    return get_logger(name, log_level, log_file)

# 使用环境变量控制
# LOG_LEVEL=DEBUG
# LOG_FILE=/var/log/myapp/app.log

"""
PyCharm 在点击"运行"时，自动做了 source 的工作
PyCharm 自动加载了，Linux 需要你手动 source .env

linux中读取环境变量过程
# 1. 写 .env 文件
echo "LOG_LEVEL=INFO" > .env

# 2. 必须执行这个命令加载
source .env

# 3. 然后才能运行 Python
python app.py  # ✅ 能读到 LOG_LEVEL
"""

优化2：不同模块不同日志文件

# 业务日志
biz_logger = get_logger("business", logging.INFO, "logs/business.log")

# 错误日志（单独记录ERROR级别）
error_logger = get_logger("error", logging.ERROR, "logs/error.log")

# 访问日志
access_logger = get_logger("access", logging.INFO, "logs/access.log")

优化3：JSON格式输出（适合日志收集系统）

import json
import logging

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno
        }
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

# 使用JSON格式
formatter = JSONFormatter()

"""
输出:
{"timestamp": "2024-01-15 10:30:45,123", "level": "INFO",
 "logger": "app", "message": "用户登录成功", "module": "test", "line": 20}
"""

五、常见问题与解决方案

Q1：日志重复输出

原因： 多次调用get_logger导致重复添加handler
解决： 代码中已有防护机制 if logger.handlers: return logger

Q2：中文乱码

原因： 文件编码问题
解决： 设置 encoding='utf-8'

Q3：日志文件权限错误

原因： 进程无写入权限
解决：

# 创建目录时设置权限
os.makedirs(log_dir, mode=0o755, exist_ok=True)

Q4：多进程写同一日志文件

问题： RotatingFileHandler不支持多进程
解决方案：

from logging.handlers import QueueHandler, QueueListener
from multiprocessing import Queue

log_queue = Queue(-1)
queue_handler = QueueHandler(log_queue)
listener = QueueListener(log_queue, file_handler, console_handler)
listener.start()

六、最佳实践总结

实践要点	说明
✅ 使用__name__作为logger名称	自动体现模块路径
✅ 设置合理的轮转大小	10-100MB为宜
✅ 生产环境使用INFO级别	DEBUG会影响性能
✅ 异常使用logger.exception()	自动记录堆栈信息
✅ 避免在循环中记录大量日志	影响性能
✅ 敏感信息脱敏	密码、token等不要记录

七、性能对比测试

import time

# 测试1：（设置级别为CRITICAL）
logger.setLevel(logging.CRITICAL)
start = time.time()
for i in range(100000):
    logger.debug(f"测试消息 {i}")  # 不输出，但有判断开销
print(f"禁用耗时: {time.time() - start:.3f}秒")

# 测试2：（INFO级别）
logger.setLevel(logging.INFO)
start = time.time()
for i in range(100000):
    logger.debug(f"测试消息 {i}")  # 不输出，但有判断开销
print(f"判断开销: {time.time() - start:.3f}秒")

# 测试3：先判断再记录（最佳实践）
logger.setLevel(logging.INFO)
start = time.time()
for i in range(100000):
    if logger.isEnabledFor(logging.DEBUG):
        logger.debug(f"测试消息 {i}")
print(f"优化后: {time.time() - start:.3f}秒")

"""
logger.debug(f"...") 无论什么级别，f-string 都会先执行。
用 isEnabledFor 判断可以避免 f-string 的执行开销。
"""

八、直接可用的完整模板

"""
production_logger.py - 生产环境日志封装
"""
import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
import os
from typing import Optional

class ProductionLogger:
    """生产级日志管理器"""
    
    _instances = {}
    
    @classmethod
    def get_logger(cls, name: str, log_level: Optional[int] = None, 
                   log_file: Optional[str] = None):
        """
        获取配置好的日志器
        
        Args:
            name: 日志器名称，建议使用__name__
            log_level: 日志级别，默认从环境变量LOG_LEVEL读取，未设置则INFO
            log_file: 日志文件路径，默认从环境变量LOG_FILE读取，未设置则logs/app.log
        
        Returns:
            配置好的Logger实例
        """
        # 配置默认值
        if log_level is None:
            log_level = getattr(logging, os.getenv('LOG_LEVEL', 'INFO'))
        if log_file is None:
            log_file = os.getenv('LOG_FILE', 'logs/app.log')
        
        # 获取或创建logger
        logger = logging.getLogger(name)
        
        # 避免重复配置
        if name in cls._instances:
            return logger
        
        logger.setLevel(log_level)
        
        # 创建日志目录
        log_dir = os.path.dirname(log_file)
        if log_dir and not os.path.exists(log_dir):
            os.makedirs(log_dir, exist_ok=True)
        
        # 文件处理器（按大小轮转）
        file_handler = RotatingFileHandler(
            log_file, 
            maxBytes=10*1024*1024,  # 10MB
            backupCount=5,
            encoding='utf-8'
        )
        file_handler.setLevel(log_level)
        
        # 控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setLevel(log_level)
        
        # 格式化器
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
        )
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        
        # 添加处理器
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)
        
        # 缓存实例
        cls._instances[name] = logger
        
        return logger

# 使用示例
if __name__ == '__main__':
    logger = ProductionLogger.get_logger(__name__)
    logger.info("日志系统初始化完成")
    logger.warning("这是一条警告")
    
    try:
        result = 10 / 0
    except Exception as e:
        logger.exception("捕获到异常")

测试执行结果:

"""
2026-04-08 16:04:09,287 - __main__ - INFO - pp.py:84 - 日志系统初始化完成
2026-04-08 16:04:09,288 - __main__ - WARNING - pp.py:85 - 这是一条警告
2026-04-08 16:04:09,289 - __main__ - ERROR - pp.py:90 - 捕获到异常
Traceback (most recent call last):
  File "D:\LLM\RAG\demo\d_tradtional_rag\utils\pp.py", line 88, in <module>
    result = 10 / 0
             ~~~^~~
ZeroDivisionError: division by zero
"""

这份代码封装了生产环境日志的所有核心要素，可直接用于实际项目。通过本文的逐行解析和示例，相信你已经完全掌握了生产级日志系统的实现原理和最佳实践。