Python 基础体系 · 第 48/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python 日志工程：Logger、Handler、结构化字段、上下文和轮转

日志不是“把字符串打印出来”，而是一条从事件产生、筛选、补充字段、格式化，到最终输出和生命周期管理的处理链。

在 Python 3.14 中，标准库 logging 将这条链拆成多个职责明确的组件：

业务代码
   │
   ▼
Logger ── 创建 LogRecord、执行日志级别和 Logger 过滤
   │
   ▼
父 Logger / root Logger（可能沿层级传播）
   │
   ▼
Handler ── 决定输出到哪里、按什么级别接收
   │
   ├── Filter ── 丢弃记录或补充字段
   │
   └── Formatter ── 把 LogRecord 转换为文本或 JSON
   │
   ▼
控制台、文件、轮转文件、队列、网络等目的地

Logger、Handler 和 Formatter 不是同一个概念：前者表示“谁产生了事件”，中者表示“事件送往哪里”，后者表示“事件最终长什么样”。LogRecord 则是这几个组件之间传递的数据对象。Python 官方文档将 LogRecord 定义为包含日志事件全部相关信息的对象，它由 Logger 在记录日志时自动创建。(docs.python.org)

一、先建立日志事件的完整模型

一次日志调用可以抽象成：

E=(n,l,m,a,x,c,s)E = (n, l, m, a, x, c, s)

其中：

  • nn：Logger 名称，例如 myapp.orders；
  • ll：日志级别，例如 INFO、ERROR；
  • mm：消息模板，例如 "order %s created"；
  • aa：模板参数，例如 ("o-1001",)；
  • xx：异常信息，可由 exc_info=True 提供；
  • cc：上下文字段，例如 request_id、user_id；
  • ss：调用位置，例如文件名、行号、函数名。

这些输入首先被封装为 LogRecord。对于：

logger.info("order %s created", order_id)

日志消息并不是立刻通过字符串拼接得到的。msg 保存模板，args 保存参数，最终消息通常通过：

record.getMessage()

计算：

msg % args

因此应优先使用参数化日志，而不是提前格式化：

# 推荐：只有真正需要输出时才格式化
logger.debug("payload size=%d", len(payload))

# 不推荐：即使 DEBUG 被关闭，也会先执行格式化
logger.debug(f"payload size={len(payload)}")

更准确地说，参数化日志主要避免字符串插值本身的开销；len(payload) 这样的参数表达式仍然会先执行。如果参数计算很重，应先检查级别：

if logger.isEnabledFor(logging.DEBUG):
    logger.debug("decoded payload=%r", expensive_decode(raw))

日志级别的数值顺序

标准级别本质上是整数：

DEBUG    10
INFO     20
WARNING  30
ERROR    40
CRITICAL 50

日志是否通过某个级别判断，可写成：

enabled⁡(record,threshold)  ⟺  record.levelno≥threshold\operatorname{enabled}(record, threshold) \iff record.levelno \ge threshold

所以级别是“最低接收级别”，不是“只接收这个级别”。

例如：

logger.setLevel(logging.INFO)

表示 INFO、WARNING、ERROR 和 CRITICAL 可以继续处理，而 DEBUG 被忽略。

还可以自定义级别：

logging.addLevelName(25, "NOTICE")

但这只增加名称映射，不会自动为 Logger 创建 notice() 方法，也不会让外部日志系统理解这个级别。除非确有协议兼容需求，否则不应随意扩展标准级别。

二、Logger：事件的命名空间和入口

1. Logger 名称形成树状层级

通常每个模块使用：

import logging

logger = logging.getLogger(__name__)

如果模块路径为：

myapp/orders/service.py

那么 __name__ 可能是：

myapp.orders.service

Logger 名称由点号分隔，形成层级：

myapp
└── myapp.orders
    └── myapp.orders.service

父子关系用于日志传播和统一配置，但它不是 Python 包导入机制的简单复制。Logger 是否存在对应的父 Logger，由 logging 的名称管理器处理。

2. Logger 和 Handler 的级别各自负责什么

日志调用至少经历两个级别判断：

Logger 级别：事件能否继续向 Handler 发送
Handler 级别：该 Handler 是否实际输出

可以形式化为：

emit⁡h(r)=(r.levelno≥Llogger)∧(r.levelno≥Lh)\operatorname{emit}_h(r) = (r.levelno \ge L_{\text{logger}}) \land (r.levelno \ge L_h)

假设：

logger.setLevel(logging.DEBUG)
console.setLevel(logging.INFO)
file_handler.setLevel(logging.DEBUG)

则：

事件 Logger 判断 控制台 文件
DEBUG 通过 丢弃 输出
INFO 通过 输出 输出
ERROR 通过 输出 输出

如果 Logger 级别设为 INFO，那么 DEBUG 在到达任何 Handler 前就已经被丢弃，Handler 不可能“把它救回来”。

Python HOWTO 明确区分了这两个级别：Logger 的级别决定哪些消息会分派给 Handler，而 Handler 的级别决定哪些消息由该 Handler 发出。(docs.python.org)

3. NOTSET 和有效级别

Logger 的级别可以是 NOTSET。对非 root Logger 来说，NOTSET 通常意味着向父 Logger 查找第一个实际设置的级别，这个过程一直向上直到 root Logger。

因此，某个 Logger 的有效级别不是简单读取：

logger.level

而应使用：

logger.getEffectiveLevel()

示例：

parent = logging.getLogger("myapp")
child = logging.getLogger("myapp.orders")

parent.setLevel(logging.WARNING)
child.setLevel(logging.NOTSET)

print(child.level)                 # 0，即 NOTSET
print(child.getEffectiveLevel())   # 30，即 WARNING

4. propagate 和重复输出

当子 Logger 处理完一条记录后，如果：

logger.propagate is True

记录会继续交给父 Logger 的 Handler，直到 root Logger 或某个 Logger 设置：

logger.propagate = False

常见重复输出场景：

app_logger.addHandler(console)
root_logger.addHandler(console)

同时 app_logger.propagate 保持默认的 True。一条日志可能先由 app_logger 的 console 输出，再传播到 root 的 console 再输出一次。

如果应用明确在顶层 Logger 安装 Handler，通常可以：

app_logger.propagate = False

但不要机械地对所有 Logger 设置 propagate=False。层级传播本身是统一配置的主要机制。

5. root Logger 与库代码

应用程序可以配置 root Logger，但库代码不应擅自调用：

logging.basicConfig(...)

也不应默认把日志写入文件或网络。库通常这样写：

# library_package/__init__.py
import logging

logging.getLogger(__name__).addHandler(logging.NullHandler())

NullHandler 不执行格式化和输出，适合让库在没有应用配置时保持安静，同时允许应用程序通过 Logger 层级接管日志。(docs.python.org)

三、Handler：把日志送往目的地

Handler 表示日志输出目的地以及与目的地相关的处理策略。常用实现包括：

Handler 用途
StreamHandler 输出到控制台或任意文本流
FileHandler 写入文件
RotatingFileHandler 按文件大小轮转
TimedRotatingFileHandler 按时间轮转
WatchedFileHandler 监视外部程序对文件的替换
QueueHandler 把记录放入队列
QueueListener 从队列取出记录，交给真正的 Handler
NullHandler 不输出，主要供库使用

应用代码一般不直接实例化基类 Handler，而是使用具体子类。官方 HOWTO 将 Handler 描述为定义处理器接口和默认行为的基类。(docs.python.org)

1. Handler 不是“日志级别的副本”

一个应用可以把同一条记录发送到多个 Handler：

INFO 及以上 ──> 控制台
DEBUG 及以上 ──> 调试文件
ERROR 及以上 ──> 告警系统

这不是复制三份业务日志，而是让不同目的地拥有不同的筛选策略。

示例：

import logging
import sys

logger = logging.getLogger("myapp")
logger.setLevel(logging.DEBUG)

console = logging.StreamHandler(sys.stderr)
console.setLevel(logging.INFO)

debug_file = logging.FileHandler("debug.log", encoding="utf-8")
debug_file.setLevel(logging.DEBUG)

logger.addHandler(console)
logger.addHandler(debug_file)

这里 Logger 必须设为 DEBUG，否则 DEBUG 记录在到达 debug_file 前就会消失。

2. Handler 的锁与慢 I/O

文件写入、网络发送和邮件发送都可能是慢操作。直接在请求线程中执行，会使请求延迟包含日志 I/O 延迟。

QueueHandler 将日志记录放入队列，QueueListener 在另一个线程中取出记录并交给真实 Handler。官方文档特别指出，这种组合适合需要快速响应客户端的 Web 服务，因为 SMTP 等潜在慢操作可以移到其他线程。(docs.python.org)

sequenceDiagram
    participant R as 请求线程
    participant QH as QueueHandler
    participant Q as queue.Queue
    participant L as QueueListener线程
    participant FH as FileHandler

    R->>QH: logger.info(...)
    QH->>Q: 放入 LogRecord
    QH-->>R: 返回
    L->>Q: 取出 LogRecord
    L->>FH: format + emit
    FH-->>L: 写入文件

但队列并不意味着日志不会丢失：

  1. 队列可能有容量上限；
  2. 进程崩溃时，尚未处理的记录可能还在内存中；
  3. 队列消费者停止后，生产者可能阻塞或报错；
  4. 关闭程序时必须先停止生产，再等待监听器处理完剩余记录。

Python 3.14 中，QueueListener 支持上下文管理器，可以使用：

with QueueListener(log_queue, file_handler) as listener:
    # 业务代码
    ...

这会把监听器的启动和退出纳入上下文生命周期。(docs.python.org)

对于多进程场景，应使用 multiprocessing.Queue 等适合进程间通信的队列；官方文档明确提醒，使用 multiprocessing 时不要使用 SimpleQueue。(docs.python.org)

四、Formatter：从 LogRecord 生成输出

Formatter 负责确定日志输出的顺序、内容和表示方式。它不会决定日志送往哪里，也不负责创建 Logger。

最常见的文本格式是：

formatter = logging.Formatter(
    fmt="%(asctime)s %(levelname)s %(name)s "
        "%(filename)s:%(lineno)d %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)

可用字段来自 LogRecord，例如：

  • levelname：级别名称；
  • levelno：级别数值；
  • name：Logger 名称；
  • message：执行参数替换后的消息；
  • pathname：源文件完整路径；
  • filename：文件名；
  • lineno：日志调用行号；
  • funcName：函数名；
  • process、processName：进程信息；
  • thread、threadName：线程信息；
  • created：创建时间的 Unix 时间戳；
  • exc_info、exc_text：异常信息。

如果格式字符串包含不存在的字段，格式化阶段可能失败。因此这段代码有风险：

formatter = logging.Formatter("%(request_id)s %(message)s")
logger.info("hello")

如果该记录没有 request_id，输出可能触发格式化错误。结构化字段应通过统一机制注入，而不是要求每个调用点手工传齐所有字段。

五、结构化日志：字段不是拼接到消息里的文本

1. 什么是结构化字段

结构化日志把事件拆成固定字段：

{
  "timestamp": "2026-09-01T10:20:30.123+00:00",
  "level": "INFO",
  "logger": "myapp.orders",
  "message": "order created",
  "fields": {
    "order_id": "o-1001",
    "amount": 39.9
  }
}

非结构化写法通常是：

2026-09-01 18:20:30 INFO order=o-1001 amount=39.9 order created

人可以阅读第二种形式，但查询系统很难可靠地区分：

amount=39.9

中的数字、字符串、缺失值和转义字符。

结构化字段的核心不是“使用 JSON”四个字，而是让字段保持独立的数据类型和稳定语义：

logger.info(
    "order created",
    extra={
        "fields": {
            "order_id": "o-1001",
            "amount": 39.9,
            "retry": False,
        }
    },
)

这里的 extra 会把额外属性放到 LogRecord 上。工程上建议把业务字段集中放在 record.fields 中，而不是把任意字段直接挂到记录对象上。这样可以减少和标准 LogRecord 属性冲突的概率。

2. 一个可运行的 JSON Formatter

标准库提供 json，但没有一个通用的内置 JSON 日志 Formatter。因此可以实现一个小型 Formatter：

import json
import logging
from datetime import datetime, timezone


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        event = {
            "timestamp": datetime.fromtimestamp(
                record.created, tz=timezone.utc
            ).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "source": {
                "file": record.filename,
                "line": record.lineno,
                "function": record.funcName,
            },
        }

        fields = getattr(record, "fields", None)
        if fields:
            event["fields"] = fields

        if record.exc_info:
            event["exception"] = {
                "type": record.exc_info[0].__name__,
                "message": str(record.exc_info[1]),
                "traceback": self.formatException(record.exc_info),
            }

        return json.dumps(
            event,
            ensure_ascii=False,
            separators=(",", ":"),
            default=str,
        )

几个细节决定了这个 Formatter 是否可靠：

  • 使用 record.getMessage()，而不是直接使用 record.msg，这样才能应用 % 参数；
  • record.created 是时间戳，需要明确转换为时区；
  • 异常不能只记录 str(exception)，否则会丢失堆栈；
  • default=str 可以处理少量非 JSON 原生类型，但也可能掩盖字段类型错误；
  • ensure_ascii=False 便于直接查看中文，但下游系统是否支持 UTF-8 仍需确认。

default=str 不适合无条件使用在严格的数据协议中。例如 Decimal 被转换成字符串后，下游可能不再把它当作数值。生产系统应对字段类型建立约束，而不是依赖 Formatter 自动兜底。

六、上下文：让请求标识自动进入每条日志

1. 上下文和业务字段的区别

业务字段描述当前事件：

order_id、amount、operation

上下文字段描述当前执行范围：

request_id、trace_id、user_id、tenant_id

同一个 HTTP 请求中的几十条日志通常都需要 request_id。如果每一次调用都手写：

logger.info("query started", extra={"fields": {"request_id": request_id}})

很容易漏写，且函数签名会被上下文污染。

contextvars 提供了上下文局部状态，适合异步任务和线程并发。官方文档说明，Context Variable 用于保存 context-local state；与带状态的上下文管理器相比，它能避免并发代码之间的状态泄漏。(docs.python.org)

2. 使用 ContextVar 保存请求上下文

from contextvars import ContextVar
from typing import Any

request_context: ContextVar[dict[str, Any]] = ContextVar(
    "request_context",
    default={},
)

设置上下文时必须保存并恢复旧值：

token = request_context.set({
    "request_id": "req-001",
    "user_id": "u-42",
})

try:
    logger.info("handling request")
finally:
    request_context.reset(token)

Python 3.14 中，ContextVar.set() 返回的 token 支持上下文管理器，因此可以写成：

with request_context.set({
    "request_id": "req-001",
    "user_id": "u-42",
}):
    logger.info("handling request")

退出 with 后，旧上下文会自动恢复。这个 Token 上下文管理器能力是 Python 3.14 新增的。(docs.python.org)

3. 使用 Filter 注入上下文

Filter 可以附加在 Logger 或 Handler 上。它不仅能判断记录是否通过，还能修改记录；Python 3.12 起，Filter 还可以返回一个新的 LogRecord，从而避免修改同一记录影响其他 Handler。(docs.python.org)

import copy
import logging


class ContextFilter:
    def filter(self, record: logging.LogRecord):
        copied = copy.copy(record)

        current = request_context.get()
        original_fields = getattr(record, "fields", {})

        copied.fields = {
            **current,
            **original_fields,
        }
        return copied

这里的覆盖顺序是：

上下文默认字段
    ↓
调用点显式传入的业务字段

因此调用点可以覆盖同名字段：

logger.info(
    "acting as system",
    extra={"fields": {"user_id": "system"}},
)

通常不应允许普通业务代码覆盖安全敏感字段，例如真实租户标识或认证主体。可以在 Filter 中对这些字段执行保护策略。

4. Logger Filter 和 Handler Filter 的边界

这是一个经常被误解的地方。

如果 Filter 挂在某个 Logger 上，它只会在该 Logger 自己调用 debug()、info() 等方法时执行。子 Logger 传播上来的记录，不会自动经过父 Logger 的 Filter。

例如：

myapp.orders.service  ──传播──> myapp

挂在 myapp Logger 上的 Filter，不一定会过滤 myapp.orders.service 产生的记录。

如果需要对最终输出统一补充字段，通常把 Filter 挂到 Handler 上更直接。Handler Filter 会在该 Handler 输出前处理记录。官方文档明确区分了 Logger Filter 和 Handler Filter 的调用时机。(docs.python.org)

七、一个完整的应用配置

下面的示例具备：

  • 控制台文本输出；
  • 文件 JSON 输出；
  • 请求上下文；
  • 业务结构化字段；
  • 异常堆栈；
  • 按文件大小轮转；
  • 关闭时安全释放 Handler。
from __future__ import annotations

import copy
import json
import logging
import logging.handlers
import sys
from contextvars import ContextVar
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


request_context: ContextVar[dict[str, Any]] = ContextVar(
    "request_context",
    default={},
)


class ContextFilter:
    def filter(self, record: logging.LogRecord):
        copied = copy.copy(record)

        context_fields = request_context.get()
        event_fields = getattr(record, "fields", {})

        copied.fields = {
            **context_fields,
            **event_fields,
        }
        return copied


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        event: dict[str, Any] = {
            "timestamp": datetime.fromtimestamp(
                record.created,
                timezone.utc,
            ).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "source": {
                "file": record.filename,
                "line": record.lineno,
                "function": record.funcName,
            },
        }

        fields = getattr(record, "fields", None)
        if fields:
            event["fields"] = fields

        if record.exc_info:
            event["exception"] = {
                "type": record.exc_info[0].__name__,
                "message": str(record.exc_info[1]),
                "traceback": self.formatException(record.exc_info),
            }

        return json.dumps(
            event,
            ensure_ascii=False,
            separators=(",", ":"),
            default=str,
        )


def configure_logging(log_dir: str = "logs") -> logging.Logger:
    path = Path(log_dir)
    path.mkdir(parents=True, exist_ok=True)

    logger = logging.getLogger("myapp")
    logger.setLevel(logging.DEBUG)
    logger.propagate = False

    # 防止配置函数被重复调用时叠加 Handler
    for handler in logger.handlers[:]:
        handler.close()
        logger.removeHandler(handler)

    context_filter = ContextFilter()

    console = logging.StreamHandler(sys.stderr)
    console.setLevel(logging.INFO)
    console.setFormatter(logging.Formatter(
        "%(asctime)s %(levelname)s %(name)s %(message)s",
        datefmt="%Y-%m-%dT%H:%M:%S%z",
    ))
    console.addFilter(context_filter)

    file_handler = logging.handlers.RotatingFileHandler(
        path / "app.log",
        maxBytes=10 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
        delay=True,
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(JsonFormatter())
    file_handler.addFilter(context_filter)

    logger.addHandler(console)
    logger.addHandler(file_handler)

    return logger


def handle_request(logger: logging.Logger) -> None:
    with request_context.set({
        "request_id": "req-001",
        "user_id": "u-42",
        "tenant_id": "tenant-a",
    }):
        logger.info("request started")

        logger.info(
            "order created",
            extra={
                "fields": {
                    "order_id": "o-1001",
                    "amount": 39.9,
                }
            },
        )

        try:
            1 / 0
        except ZeroDivisionError:
            logger.exception(
                "order calculation failed",
                extra={
                    "fields": {
                        "order_id": "o-1001",
                        "operation": "calculate_total",
                    }
                },
            )


if __name__ == "__main__":
    logger = configure_logging()
    handle_request(logger)

    for handler in logger.handlers:
        handler.flush()
        handler.close()

运行：

python app.py

控制台预期类似：

2026-09-01T18:20:30+0800 INFO myapp request started
2026-09-01T18:20:30+0800 INFO myapp order created
2026-09-01T18:20:30+0800 ERROR myapp order calculation failed
Traceback (most recent call last):
  ...
ZeroDivisionError: division by zero

文件中的记录则是单行 JSON，例如：

{"timestamp":"2026-09-01T10:20:30.123+00:00","level":"INFO","logger":"myapp","message":"order created","source":{"file":"app.py","line":100,"function":"handle_request"},"fields":{"request_id":"req-001","user_id":"u-42","tenant_id":"tenant-a","order_id":"o-1001","amount":39.9}}

这里控制台和文件看到的是同一个事件，但使用了不同 Formatter。控制台优先可读性，文件优先机器查询能力。

示例中调用 logger.exception() 的位置位于 except 块内，因此 LogRecord.exc_info 可以携带当前异常。若在异常处理块外调用：

logger.exception("something failed")

通常不会得到想要的当前异常堆栈，应该使用：

logger.error("something failed", exc_info=exc)

或者显式传入：

logger.error("something failed", exc_info=True)

但 exc_info=True 只有在当前线程仍处于异常处理上下文中时才有意义。

八、轮转：控制日志文件的生命周期

如果使用普通 FileHandler：

handler = logging.FileHandler("app.log")

文件会持续增长。轮转的目标是把一个无限增长的文件转换为一组有界文件：

app.log
app.log.1
app.log.2
app.log.3

1. 按大小轮转：RotatingFileHandler

handler = logging.handlers.RotatingFileHandler(
    "app.log",
    maxBytes=10 * 1024 * 1024,
    backupCount=5,
    encoding="utf-8",
)

其基本流程可以抽象为：

if current_size+incoming_record>maxBytes\text{if } current\_size + incoming\_record > maxBytes

则：

1. 关闭当前 app.log
2. app.log.4 改名为 app.log.5
3. app.log.3 改名为 app.log.4
4. ...
5. app.log 改名为 app.log.1
6. 新建 app.log
7. 写入当前记录

当 maxBytes 或 backupCount 任意一个为零时，轮转不会发生；要启用轮转，通常必须同时设置非零的 maxBytes 和 backupCount。官方文档还规定，当前文件始终使用基础文件名，旧文件依次使用 .1、.2 等后缀。(docs.python.org)

风险在于：轮转检查发生在写入过程中，不是后台定时任务。若应用长时间没有新日志，即使时间已经过去，文件也不会因为“到点”自动变化。

2. 按时间轮转：TimedRotatingFileHandler

handler = logging.handlers.TimedRotatingFileHandler(
    "app.log",
    when="midnight",
    interval=1,
    backupCount=14,
    encoding="utf-8",
    utc=True,
)

常用 when 包括：

S          秒
M          分钟
H          小时
D          天
W0-W6      按星期轮转
midnight   午夜轮转

时间轮转同样通常在下一次 emit() 时触发，而不是由独立定时线程保证准点执行。utc=True 可以让轮转时间基于 UTC，但业务团队必须明确文件名时间和日志事件时间采用什么时区。官方文档规定了 when、interval、utc 和 atTime 的组合语义。(docs.python.org)

3. 外部轮转：WatchedFileHandler

Linux 环境常由 logrotate 或类似工具负责：

app.log
    ↓
外部工具重命名为 app.log.1
    ↓
创建新的 app.log

此时进程仍可能持有旧文件描述符。如果继续向旧描述符写入，新建的 app.log 就收不到日志。

WatchedFileHandler 会检查文件的设备号和 inode；发现文件被替换后，关闭旧流并按文件名重新打开。它适用于 Unix/Linux，不适合 Windows，因为 Windows 下打开的文件通常不能被正常移动或重命名，官方文档也明确说明了这一平台边界。(docs.python.org)

三种方案的选择逻辑是：

场景 方案
应用单独管理日志文件 RotatingFileHandler
希望按时间切分 TimedRotatingFileHandler
Unix/Linux 上由 logrotate 管理 WatchedFileHandler
容器环境由运行时收集标准输出 StreamHandler

不要同时让应用内轮转和 logrotate 操作同一个文件。两个机制都可能重命名文件，最终会出现文件遗漏、日志分散或轮转顺序异常。

4. 多进程写同一个轮转文件

标准 RotatingFileHandler 适合单进程内的线程并发，但不应简单假设多个进程共享它就能安全完成轮转。

多个进程可能同时判断：

当前文件接近 maxBytes

然后同时执行改名操作。可能出现：

  • 一个进程覆盖另一个进程的轮转结果；
  • 文件名序号不连续；
  • 记录交错或丢失；
  • 某个进程继续写入已经被改名的文件。

标准库文档和 Logging Cookbook 对多进程文件日志存在限制，推荐将日志集中到单独的监听进程或使用专门的日志收集系统；QueueHandler 和 QueueListener 可以把多个生产者与单一输出端解耦，但跨进程时必须正确配置进程队列和关闭流程。(docs.python.org)

九、用 dictConfig 管理环境差异

当日志配置包含多个 Handler、Formatter 和 Filter 时，直接写 Python 初始化代码容易与环境配置混在一起。logging.config.dictConfig() 可以使用字典描述 Formatter、Handler、Logger 之间的连接关系。配置字典的 version 当前有效值为 1。(docs.python.org)

一个基础配置如下：

import logging.config


LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "console": {
            "format": (
                "%(asctime)s %(levelname)s "
                "%(name)s %(message)s"
            ),
        },
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "console",
            "stream": "ext://sys.stderr",
        },
    },
    "loggers": {
        "myapp": {
            "level": "DEBUG",
            "handlers": ["console"],
            "propagate": False,
        },
    },
}

logging.config.dictConfig(LOGGING)

disable_existing_loggers 的默认值是 True。这意味着如果配置中省略它，已有的非 root Logger 可能被禁用。应用依赖第三方库时，通常应显式写出：

"disable_existing_loggers": False

否则你可能看到：

自己的应用日志正常
第三方库日志突然消失

dictConfig 支持通过 "class" 和 "()" 导入或创建对象，因此配置文件不是纯数据格式。官方文档提醒，应把不可信配置当作高风险输入，因为配置机制可能导入并调用用户指定的对象。(docs.python.org)

环境差异可以放在配置生成阶段，而不是把 Secret 写入日志配置：

import os

level = os.getenv("LOG_LEVEL", "INFO").upper()

但不要直接信任环境变量：

"level": os.getenv("LOG_LEVEL", "INFO")

如果值是：

DEBUG
INFO
WARNING
ERROR
CRITICAL

可以接受；如果是拼写错误：

INF0

应在启动阶段快速失败，而不是让程序启动后才发现日志没有输出。

VALID_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}

def parse_log_level(value: str) -> str:
    normalized = value.strip().upper()
    if normalized not in VALID_LEVELS:
        raise ValueError(
            f"invalid LOG_LEVEL={value!r}; "
            f"expected one of {sorted(VALID_LEVELS)}"
        )
    return normalized

日志级别配置通常不是 Secret，但日志文件路径、远程日志地址、认证令牌等配置可能涉及敏感信息。尤其不要把环境变量中的密码、数据库连接串或 API Token 自动放进 fields。

十、异常日志：记录故障路径，而不是只记录错误文本

下面两种日志信息的诊断价值不同：

logger.error("request failed")

和：

logger.exception(
    "request failed",
    extra={"fields": {"request_id": request_id}},
)

第一种只能说明“失败发生了”。第二种还提供：

  • 异常类型；
  • 异常消息；
  • 调用堆栈；
  • 请求上下文；
  • 业务操作字段。

推荐在能够处理异常、或者需要转换异常边界的位置记录一次完整堆栈：

try:
    result = call_dependency()
except TimeoutError:
    logger.exception(
        "dependency timeout",
        extra={"fields": {"dependency": "payment"}},
    )
    raise

不要在每一层都 logger.exception() 后继续抛出，否则同一个异常会被记录多次：

repository failed
service failed
controller failed

如果每层都带完整堆栈，单次故障可能制造大量重复日志。更合理的策略是：

底层：补充上下文后继续抛出
边界层：记录一次完整堆栈并转换响应

但如果底层正在把异常转换成完全不同的故障语义，例如把第三方异常转换成领域异常，可以在转换点记录必要信息，不过仍需控制重复程度。

十一、常见失败表现与诊断方法

1. “我设置了 Handler 的 DEBUG，但 DEBUG 没有输出”

检查顺序：

print(logger.level)
print(logger.getEffectiveLevel())
print(handler.level)
print(logger.handlers)
print(logger.propagate)

重点判断：

Logger 有效级别是否已经高于 DEBUG？
Handler 是否真的挂在这条 Logger 路径上？
是否配置了错误的 Logger 名称？
是否被 dictConfig 禁用了？

Handler 不能恢复已经被 Logger 丢弃的记录。

2. “同一条日志输出两次”

检查：

for current in [logger, logging.getLogger(),]:
    print(
        current.name,
        current.handlers,
        current.propagate,
    )

常见原因是：

子 Logger 有 Handler
同时 root Logger 也有 Handler
且子 Logger propagate=True

修复方式通常是二选一：

只在 root 配置 Handler，让子 Logger 传播

或者：

只在应用顶层 Logger 配置 Handler，并设置 propagate=False

3. “JSON 日志偶尔格式化失败”

如果使用 %() 格式字符串，检查所有记录是否都有对应属性：

"%(request_id)s %(message)s"

而自定义字段未注入时会失败。

JSON Formatter 则应避免直接索引：

record.request_id

更安全的是：

getattr(record, "request_id", None)

或者统一使用：

getattr(record, "fields", {})

4. “上下文串到了下一个请求”

错误代码：

request_context.set({"request_id": request_id})
handle_request()
# 没有恢复旧值

在复用线程、异步任务或异常路径中，这会导致后续代码看到错误的请求标识。必须使用 token 恢复：

token = request_context.set(context)
try:
    handle_request()
finally:
    request_context.reset(token)

Python 3.14 可使用：

with request_context.set(context):
    handle_request()

5. “日志轮转后仍然写到旧文件”

如果轮转由外部 logrotate 完成，而应用使用普通 FileHandler，进程可能仍持有旧文件描述符。Unix/Linux 下使用 WatchedFileHandler，或者改为输出标准错误并交给运行环境收集。Windows 下不要照搬 WatchedFileHandler 方案。(docs.python.org)

6. “日志本身拖慢了请求”

检查是否存在以下路径：

请求线程
  └── Formatter 复杂序列化
      └── 文件写入
          └── 网络或磁盘阻塞

对高延迟目的地使用 QueueHandler；但要同时观察队列长度、消费者状态和关闭时的排空行为。队列只是改变等待位置，不会消除 I/O 成本。

十二、哪些字段值得进入结构化日志

一个字段应当满足至少一个条件：

  1. 能帮助定位请求或调用链；
  2. 能帮助按业务对象聚合故障；
  3. 能帮助区分重试、超时和依赖失败；
  4. 能帮助关联日志、指标和 Trace；
  5. 能帮助恢复故障发生时的系统状态。

常见字段包括：

request_id
trace_id
span_id
user_id
tenant_id
service
environment
operation
dependency
retry_count
duration_ms
status_code
error_type

字段命名必须稳定。下面两种写法会增加查询复杂度：

{"requestId":"req-1"}
{"request_id":"req-2"}

日志字段也不应无限增长。把完整请求体、响应体和用户输入全部写入日志会带来：

  • 隐私泄露；
  • 日志体积膨胀；
  • 序列化开销；
  • 轮转频率增加；
  • 下游索引成本增加。

对于密码、Token、Cookie、身份证号、银行卡号等敏感数据，应在进入日志前脱敏或禁止记录。尤其不要依赖 Formatter 在最后一步“猜测哪些内容敏感”，因为敏感数据可能嵌套在任意对象的字符串表示中。

十三、与指标、Trace 和故障定位的边界

日志、指标和 Trace 解决的问题不同：

日志：某个具体事件发生了什么？
指标：一段时间内系统整体表现如何？
Trace：一次请求跨越了哪些服务和调用？

例如：

日志：
payment timeout dependency=payment request_id=req-1

指标：
payment_request_timeout_total += 1
payment_request_latency_seconds.observe(2.4)

Trace：
request span
  └── payment.client span
       └── timeout event

不要只用日志统计高频数值。大量日志用于聚合请求量、延迟和错误率，通常不如指标直接；也不要只用指标定位单次异常，因为指标一般缺少具体业务对象和堆栈。

日志上下文应尽量与 Trace 上下文共享标识：

request_id
trace_id
span_id

这样可以从指标告警进入 Trace，再从 Trace 进入具体日志，而不是在不同系统中凭时间戳猜测关联关系。

十四、最终的组件边界

可以用下面的判断确认设计是否清晰：

Logger
  负责：命名、创建记录、Logger 级别、传播

LogRecord
  负责：承载事件数据、消息模板、调用位置、异常和额外字段

Filter
  负责：细粒度放行/拒绝，或补充上下文

Handler
  负责：选择输出目的地、Handler 级别、输出生命周期

Formatter
  负责：将记录表示为文本、JSON 或其他格式

Rotating Handler
  负责：控制文件按大小或时间切分

ContextVar
  负责：在当前异步任务或线程上下文中保存请求范围状态

QueueHandler / QueueListener
  负责：把日志生产和慢速输出解耦

日志工程真正需要解决的不是“打印哪句话”，而是保证这条数据流在并发、异常、配置变化、文件增长和进程退出时仍然可解释：

事件产生
  → 级别筛选
  → 层级传播
  → 上下文与业务字段合并
  → Handler 筛选
  → 格式化
  → 输出
  → 轮转或异步消费

只要每一步的责任边界、失败表现和生命周期都明确，logging 才能从调试工具变成可用于生产故障定位的基础设施。

系列导航与关联阅读

官方资料

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。