一、为什么需要日志管理

日志是程序运行时的”黑匣子”，记录系统状态、用户行为、异常信息和性能数据。没有日志，生产环境出问题时如同盲人摸象。

# 程序崩溃，只能看报错堆栈
# 不知道请求参数是什么
# 不知道数据流转到哪一步
# 不知道性能瓶颈在哪
# 复现 Bug 靠猜测

2026-07-16 14:32:10 [INFO] 收到请求: /api/calculate, params={"a":10,"b":0}
2026-07-16 14:32:10 [DEBUG] 执行计算逻辑: divide(10, 0)
2026-07-16 14:32:10 [ERROR] 计算异常: ZeroDivisionError: division by zero
2026-07-16 14:32:10 [INFO] 返回响应: 400, {"error":"除数不能为零"}

二、日志层级（Level）深度解析

2.1 Python 标准日志级别

2.2 如何正确选择日志级别

import logging

logger = logging.getLogger(__name__)

# DEBUG：开发调试，生产不输出
logger.debug(f"函数 divide 被调用，参数: a={a}, b={b}")
logger.debug(f"数据库查询 SQL: {sql}")

# INFO：关键业务节点，生产需要
logger.info(f"用户 {user_id} 登录成功，IP: {client_ip}")
logger.info(f"订单 {order_id} 创建成功，金额: {amount}")
logger.info("服务启动完成，监听端口 8000")

# WARNING：不影响主流程但需要关注
logger.warning(f"数据库连接池使用率超过 80%: {usage}%")
logger.warning(f"请求重试第 {retry_count} 次，URL: {url}")
logger.warning("使用默认配置，未找到 .env 文件")

# ERROR：功能异常，需要排查修复
logger.error(f"支付接口调用失败: {e}", exc_info=True)
logger.error(f"订单 {order_id} 状态更新失败，当前状态: {status}")

# CRITICAL：系统级灾难
logger.critical("数据库连接全部断开，服务不可用")
logger.critical("磁盘空间不足，写入操作已中止")

2.3 日志级别过滤原理

Logger 和 Handler 各自有级别设置，消息只有同时满足两者级别才会输出。

# 示例：Logger 级别为 INFO，Handler 级别为 ERROR
# INFO 级别的日志会被 Logger 接收，但被 Handler 过滤
# ERROR 级别的日志会通过所有过滤

logger = logging.getLogger("myapp")
logger.setLevel(logging.INFO)           # 接收 INFO 及以上

handler = logging.StreamHandler()
handler.setLevel(logging.ERROR)         # 只输出 ERROR 及以上
logger.addHandler(handler)

logger.info("这条不会输出")    # 被 Handler 过滤
logger.error("这条会输出")     # 通过所有过滤

三、Python logging 模块架构

3.1 核心组件关系

Logger（记录器）
  ├── 层级命名："myapp.api.auth" 继承 "myapp.api"
  ├── 级别过滤：setLevel()
  ├── 传播开关：propagate = True/False
  └── Handler（处理器）
        ├── StreamHandler    → 输出到控制台
        ├── FileHandler      → 输出到文件
        ├── RotatingFileHandler → 按大小轮转
        ├── TimedRotatingFileHandler → 按时间轮转
        └── 级别过滤 + Formatter（格式器）
              └── "%(asctime)s [%(levelname)s] %(message)s"

3.2 Logger 层级与继承

Logger 采用点号分隔的命名空间，形成树形结构，子 Logger 继承父 Logger 的配置。

import logging

# 根 Logger
root = logging.getLogger()           # ""

# 子 Logger
app = logging.getLogger("myapp")     # 继承 root
api = logging.getLogger("myapp.api") # 继承 myapp
auth = logging.getLogger("myapp.api.auth") # 继承 myapp.api

# 配置 root，所有子 Logger 默认继承
root.setLevel(logging.WARNING)
handler = logging.StreamHandler()
root.addHandler(handler)

# myapp.api.auth 会继承 root 的 Handler 和 Level
# 除非自己 addHandler 或 setLevel

四、配置日志系统（代码/字典/文件）

4.1 代码方式配置（最基础）

import logging
import sys

def setup_logging(level: int = logging.INFO) -> None:
    """基础日志配置。"""
    formatter = logging.Formatter(
        fmt="%(asctime)s [%(levelname)-8s] %(name)s:%(lineno)d - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    # 控制台 Handler
    console = logging.StreamHandler(sys.stdout)
    console.setLevel(level)
    console.setFormatter(formatter)

    # 文件 Handler
    file_handler = logging.FileHandler("app.log", encoding="utf-8")
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    # 根 Logger
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    root.addHandler(console)
    root.addHandler(file_handler)


# 使用
setup_logging()
logger = logging.getLogger(__name__)
logger.info("应用启动")

4.2 dictConfig 方式（推荐）

import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,

    "formatters": {
        "standard": {
            "format": "%(asctime)s [%(levelname)-8s] %(name)s:%(lineno)d - %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S"
        },
        "detailed": {
            "format": "%(asctime)s [%(levelname)-8s] %(name)s %(filename)s:%(lineno)d %(funcName)s() - %(message)s"
        }
    },

    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
            "stream": "ext://sys.stdout"
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "DEBUG",
            "formatter": "detailed",
            "filename": "logs/app.log",
            "maxBytes": 10485760,    # 10MB
            "backupCount": 5,
            "encoding": "utf-8"
        },
        "error_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "ERROR",
            "formatter": "detailed",
            "filename": "logs/error.log",
            "maxBytes": 10485760,
            "backupCount": 10,
            "encoding": "utf-8"
        }
    },

    "loggers": {
        "": {  # root logger
            "level": "DEBUG",
            "handlers": ["console", "file", "error_file"],
            "propagate": False
        },
        "uvicorn": {
            "level": "INFO",
            "handlers": ["console"],
            "propagate": False
        },
        "sqlalchemy.engine": {
            "level": "WARNING",
            "handlers": ["file"],
            "propagate": False
        }
    }
}

logging.config.dictConfig(LOGGING_CONFIG)

4.3 从文件加载配置

import logging.config
import json

# JSON 配置文件
with open("logging.json", "r", encoding="utf-8") as f:
    config = json.load(f)
logging.config.dictConfig(config)

# 或通过环境变量选择配置
import os
env = os.getenv("ENV", "dev")
config_file = f"config/logging.{env}.json"
with open(config_file, "r", encoding="utf-8") as f:
    logging.config.dictConfig(json.load(f))

五、日志文件管理策略

5.1 文件轮转策略对比

5.2 按大小轮转配置

from logging.handlers import RotatingFileHandler

# 10MB 一个文件，保留 5 个备份
handler = RotatingFileHandler(
    "app.log",
    maxBytes=10 * 1024 * 1024,  # 10MB
    backupCount=5,
    encoding="utf-8"
)
# 生成: app.log, app.log.1, app.log.2, ..., app.log.5

5.3 按时间轮转配置

from logging.handlers import TimedRotatingFileHandler

# 每天午夜轮转，保留 30 天
handler = TimedRotatingFileHandler(
    "app.log",
    when="midnight",      # 可选: S(秒), M(分), H(小时), D(天), W0-W6(周几)
    interval=1,
    backupCount=30,
    encoding="utf-8"
)
# 生成: app.log, app.log.2026-07-15, app.log.2026-07-14, ...

5.4 配合 Linux logrotate

# /etc/logrotate.d/myapp
/var/log/myapp/*.log {
    daily
    rotate 30
    compress
    delaycompress
    missingok
    notifempty
    create 0644 www-data www-data
    sharedscripts
    postrotate
        # 通知应用重新打开日志文件
        kill -USR1 $(cat /var/run/myapp.pid)
    endscript
}

六、不同环境的日志配置

6.1 环境配置对照

level: DEBUG
handlers: [console]
format: 简洁，带行号
propagate: True

level: DEBUG
handlers: [console, file]
format: 详细，带函数名
propagate: False

level: INFO
handlers: [file, error_file]
format: JSON 结构化
propagate: False

level: INFO
handlers: [file]
format: JSON 结构化
propagate: False

6.2 环境切换实现

import logging.config
import os

def get_logging_config(env: str) -> dict:
    """根据环境返回日志配置。"""
    base = {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "simple": {"format": "%(levelname)s: %(message)s"},
            "standard": {"format": "%(asctime)s [%(levelname)s] %(name)s - %(message)s"},
            "json": {"class": "pythonjsonlogger.jsonlogger.JsonFormatter",
                     "format": "%(asctime)s %(levelname)s %(name)s %(message)s"}
        },
        "handlers": {},
        "loggers": {}
    }

    if env == "dev":
        base["handlers"]["console"] = {
            "class": "logging.StreamHandler",
            "level": "DEBUG",
            "formatter": "standard"
        }
        base["loggers"][""] = {"level": "DEBUG", "handlers": ["console"]}

    elif env == "prod":
        base["handlers"]["file"] = {
            "class": "logging.handlers.TimedRotatingFileHandler",
            "level": "INFO",
            "formatter": "json",
            "filename": "/var/log/myapp/app.log",
            "when": "midnight",
            "backupCount": 30
        }
        base["handlers"]["error"] = {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "ERROR",
            "formatter": "json",
            "filename": "/var/log/myapp/error.log",
            "maxBytes": 10485760,
            "backupCount": 10
        }
        base["loggers"][""] = {"level": "INFO", "handlers": ["file", "error"]}

    return base


env = os.getenv("ENV", "dev")
logging.config.dictConfig(get_logging_config(env))

七、结构化日志与 JSON 输出

7.1 普通文本 vs 结构化 JSON

2026-07-16 14:32:10 [INFO] myapp:231 - 
用户 12345 登录成功，IP: 192.168.1.1，耗时: 45ms

{"timestamp":"2026-07-16T14:32:10Z",
 "level":"INFO",
 "logger":"myapp.auth",
 "event":"user_login",
 "user_id":"12345",
 "client_ip":"192.168.1.1",
 "duration_ms":45}

7.2 python-json-logger 配置

# pip install python-json-logger

from pythonjsonlogger import jsonlogger
import logging

logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    "%(asctime)s %(levelname)s %(name)s %(message)s",
    rename_fields={"levelname": "level", "asctime": "timestamp"}
)
logHandler.setFormatter(formatter)

logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

# 输出结构化日志
logger.info("用户登录", extra={"user_id": "12345", "ip": "192.168.1.1"})
# {"timestamp": "2026-07-16 14:32:10", "level": "INFO", "name": "root",
#  "message": "用户登录", "user_id": "12345", "ip": "192.168.1.1"}

7.3 统一日志上下文（Context）

import logging
import contextvars
import uuid

# 上下文变量
request_id = contextvars.ContextVar("request_id", default="")

class ContextFilter(logging.Filter):
    """自动注入请求 ID 到日志记录。"""
    def filter(self, record):
        record.request_id = request_id.get()
        return True

# 配置
logger = logging.getLogger("myapp")
logger.addFilter(ContextFilter())

formatter = logging.Formatter(
    "%(asctime)s [%(levelname)s] [%(request_id)s] %(message)s"
)

# 使用
request_id.set(str(uuid.uuid4())[:8])
logger.info("处理请求开始")
logger.info("查询数据库")
logger.info("返回响应")
# 所有日志自动携带同一个 request_id，便于链路追踪

八、性能优化与异步日志

8.1 日志性能要点

8.2 QueueHandler 异步配置

import logging
import logging.handlers
import queue

# 创建日志队列
log_queue = queue.Queue(-1)  # 无界队列

# 队列 Handler（生产者）
queue_handler = logging.handlers.QueueHandler(log_queue)

# 实际写入 Handler（消费者）
file_handler = logging.FileHandler("app.log", encoding="utf-8")
file_handler.setFormatter(logging.Formatter(
    "%(asctime)s [%(levelname)s] %(message)s"
))

# 队列监听器，在独立线程中消费日志
listener = logging.handlers.QueueListener(
    log_queue, file_handler, respect_handler_level=True
)

# 配置根 Logger
root = logging.getLogger()
root.addHandler(queue_handler)
root.setLevel(logging.DEBUG)

# 启动监听器
listener.start()

# ... 应用运行 ...

# 关闭时停止监听器
listener.stop()

九、日志安全与敏感信息处理

9.1 敏感信息红线

9.2 敏感信息脱敏

import logging
import re

class SensitiveDataFilter(logging.Filter):
    """日志敏感信息脱敏过滤器。"""

    PATTERNS = {
        "password": re.compile(r'("password"\s*:\s*")[^"]*("', re.I),
        "token": re.compile(r'(Bearer\s+)[\w\-\.]+', re.I),
        "phone": re.compile(r'(1\d{2})\d{4}(\d{4})'),
        "id_card": re.compile(r'(\d{6})\d{8}(\d{4})'),
    }

    def filter(self, record):
        msg = record.getMessage()
        for name, pattern in self.PATTERNS.items():
            if name in ("password", "token"):
                msg = pattern.sub(r'\1***\2', msg)
            else:
                msg = pattern.sub(r'\1****\2', msg)
        record.msg = msg
        record.args = ()
        return True

# 应用过滤器
logger = logging.getLogger("myapp")
logger.addFilter(SensitiveDataFilter())

# 测试
logger.info('登录请求: {"username":"alice","password":"Secret123"}')
# 输出: 登录请求: {"username":"alice","password":"***"}

9.3 安全的日志记录实践

# ❌ 错误：记录敏感信息
logger.info(f"用户登录: {username}, 密码: {password}")
logger.debug(f"API 请求头: {headers}")  # 可能包含 Token

# ✅ 正确：脱敏后记录
logger.info(f"用户登录: {username}")
logger.debug(f"API 请求头 keys: {list(headers.keys())}")

# ✅ 正确：只记录必要的标识
logger.info(f"支付请求: order_id={order_id}, amount={amount}")
# 不要记录: card_number, cvv, 持卡人姓名

十、常见问题与排查模板

排查检查清单

# 1. 检查 Logger 层级和级别
for name in ["", "myapp", "myapp.api"]:
    logger = logging.getLogger(name)
    print(f"{name}: level={logger.level}, handlers={logger.handlers}, propagate={logger.propagate}")

# 2. 检查所有 Handler
for handler in logging.getLogger().handlers:
    print(f"Handler: {handler}, level={handler.level}, formatter={handler.formatter}")

# 3. 强制输出测试
logging.getLogger().debug("debug test")
logging.getLogger().info("info test")
logging.getLogger().warning("warning test")