Python 日志一条变两条：从 handler 重复注册到传播链修复

适用场景与现象

适用于 Python 3.10 及以上的命令行程序、后台任务和自有服务。业务代码只调用一次 logger.info()，同一进程的原始输出却出现两行；或者应用工厂每执行一次，同一事件就多输出一遍。

先分清重复发生在哪一层：若进程的 stderr 只有一条，而日志平台出现两条，优先检查采集路径是否同时包含容器输出和文件。若两条记录的 process_id 不同，还应检查多个 Worker 是否都执行了同一任务。本篇处理同一进程里，一次日志调用经过多个输出路径的问题，不把日志去重当作业务幂等。

一、在独立进程里复现两种重复

把下面内容保存为 duplicate_demo.py，执行 python duplicate_demo.py。使用内存流，结果不会依赖终端或采集器。

import io
import logging


def make_handler(stream: io.StringIO) -> logging.StreamHandler:
    """创建独立输出器，用于复现相同目标上的重复写入。"""
    handler = logging.StreamHandler(stream)
    handler.setFormatter(logging.Formatter("%(name)s %(message)s"))
    return handler


def main() -> None:
    """隔离两种故障，避免把传播重复和注册重复混在一起。"""
    stream = io.StringIO()
    root = logging.getLogger()
    logger = logging.getLogger("demo.orders")
    logger.setLevel(logging.INFO)
    logger.propagate = True
    logging.basicConfig(handlers=[make_handler(stream)], force=True)
    child_handler = make_handler(stream)
    logger.addHandler(child_handler)
    logger.info("订单校验完成")
    print("父子同时输出行数：", len(stream.getvalue().splitlines()))

    logger.removeHandler(child_handler)
    child_handler.close()
    stream.seek(0)
    stream.truncate(0)
    logger.propagate = False
    first = make_handler(stream)
    second = make_handler(stream)
    logger.addHandler(first)
    logger.addHandler(second)
    logger.info("订单校验完成")
    print("重复注册输出行数：", len(stream.getvalue().splitlines()))

    for handler in (first, second):
        logger.removeHandler(handler)
        handler.close()
    for handler in list(root.handlers):
        root.removeHandler(handler)
        handler.close()


if __name__ == "__main__":
    main()

两次都应显示 2。第一种是子 logger 与根 logger 各写一次；第二种没有向父级传播，两个不同 handler 仍各写一次。反复调用 getLogger("demo.orders") 本身不会创建新的 logger，问题通常出在每次初始化都创建新的 handler。Python logging API

二、查看真实传播路径，不只看 handler 数量

在发生重复的进程内运行下面的诊断函数。另开 Python 进程只能看到新进程的配置，不能诊断正在运行的服务。

import json
import logging


def inspect_logging_path(logger_name: str) -> None:
    """仅输出配置元数据，避免暴露日志文件路径和业务内容。"""
    current: logging.Logger | None = logging.getLogger(logger_name)
    while current is not None:
        print(json.dumps({
            "logger_name": current.name,
            "configured_level": logging.getLevelName(current.level),
            "effective_level": logging.getLevelName(current.getEffectiveLevel()),
            "is_disabled": current.disabled,
            "should_propagate": current.propagate,
            "handlers": [
                {
                    "handler_id": id(handler),
                    "handler_type": type(handler).__name__,
                    "handler_level": logging.getLevelName(handler.level),
                }
                for handler in current.handlers
            ],
        }, ensure_ascii=False))
        if not current.propagate:
            break
        current = current.parent


inspect_logging_path("service.orders")

诊断时记录一次冷启动、一次应用工厂调用以及再次调用后的结果：

观察结果	优先检查	修复方向
同一 logger 的 handler 数量持续增加	工厂、构造函数或初始化钩子中的 addHandler	把配置移到进程入口
子 logger 和祖先都有输出 handler	propagate=True 的传播链	明确由哪一级负责输出
同一个 handler_id 出现在链上多处	同一对象被挂到父子两级	只保留一个挂载位置
handler 数量稳定，原始输出仍重复	业务调用次数、Worker、采集链路	按进程与事件来源继续取证

不同 handler_id 不代表不同目标：两个 StreamHandler 可能都写 stderr。表中是定位线索，还要在受控调试环境里核对实际输出目标。

两个常见误判：

• hasHandlers() 会沿有效传播链查找祖先，不能用它判断“当前 logger 是否已安装自己的 handler”。
• 提高 root 的 logger 级别不一定阻止显式设为 INFO 的子 logger 记录；传播阶段仍要看接收 handler 的级别。不要靠调高级别掩盖重复。Python logging API

三、方案 A：自有程序由入口统一配置

对于自己控制生命周期的程序，选择 root 统一输出，业务模块只获取 logger。保存为 logging_setup.py：

import json
import logging
import sys
from typing import TextIO


class JsonFormatter(logging.Formatter):
    """输出固定字段，第三方日志不提供 event 时使用默认值。"""

    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "logger_name": record.name,
            "process_id": record.process,
            "event": getattr(record, "event", "application_log"),
            "message": record.getMessage(),
        }
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        return json.dumps(payload, ensure_ascii=False)


def configure_logging(stream: TextIO | None = None) -> None:
    """仅用于自有程序启动阶段，替换由该入口管理的根输出配置。"""
    handler = logging.StreamHandler(sys.stderr if stream is None else stream)
    handler.setLevel(logging.INFO)
    handler.setFormatter(JsonFormatter())
    logging.basicConfig(level=logging.INFO, handlers=[handler], force=True)

在入口调用一次 configure_logging()，然后启动线程或创建应用。业务模块使用 logging.getLogger(__name__)，通过 logger.info("订单校验完成", extra={"event": "order_validated"}) 写入日志，不安装 StreamHandler。

force=True 会移除并关闭 root 上原有的 handler；它不会清理子 logger 的 handler，也不是框架配置的通用补丁。Python basicConfig

迁移已有程序时，先搜索配置来源：

rg -n 'addHandler|basicConfig|dictConfig|StreamHandler|FileHandler|propagate' .

删除业务模块和工厂中的输出器注册代码，再重启验证。在启动阶段确需清理历史实例时，只处理已确认归本应用管理的 handler，先 removeHandler()，再 close()；共享 handler 要从全部挂载位置卸载后再关闭。不要遍历所有第三方 logger 批量清空。

这里的 JSON 格式器只提供诊断字段，不负责自动脱敏。不要把令牌、Cookie、原始请求体或敏感身份数据放入 message、extra 或异常文本；异常可能携带输入内容，应在业务边界处理后再记录。

四、方案 B：框架托管时在已有配置中明确归属

若 Django、ASGI 服务或任务框架负责配置，沿用它的启动配置入口。先明确框架日志与业务日志各自的路由，再合并配置。下面是独立业务命名空间的配置片段，应并入已有配置，不能覆盖完整框架配置：

BUSINESS_LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "json": {"()": "logging_setup.JsonFormatter"},
    },
    "handlers": {
        "business_console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "json",
            "stream": "ext://sys.stderr",
        },
    },
    "loggers": {
        "service": {
            "level": "INFO",
            "handlers": ["business_console"],
            "propagate": False,
        },
        "service.orders": {
            "level": "NOTSET",
            "handlers": [],
            "propagate": True,
        },
    },
}

在这种设计中，service.orders 传给 service，后者输出后停止向 root 传播。其他业务子 logger 也应移除各自的输出 handler 并传播到 service。若实际名称是 myapp.orders，要按真实命名空间改配置；__name__ 不会自动落在 service 下。

disable_existing_loggers=False 用于避免未列出的现有非根 logger 因配置而被禁用；它不承诺现有 handler 都被移除。最终仍要执行路径检查。Python logging.config

直接把所有 logger 的 propagate 改成 False 很危险：没有自身输出 handler 的 logger 可能失去 INFO 输出路径。必须同时明确它自己的输出归属。

五、用行数、字段和丢失路径做回归验证

把以下代码保存为 test_logging_setup.py，与 logging_setup.py 放在同一目录。测试应在独立进程运行，因为它会替换 root 配置。

import io
import json
import logging
import unittest

from logging_setup import configure_logging


class LoggingSetupTests(unittest.TestCase):
    """验证重复配置、层级传播与级别边界，防止修复后丢日志。"""

    def setUp(self) -> None:
        self.stream = io.StringIO()
        self.logger = logging.getLogger("practice_logging.orders")
        self.logger.setLevel(logging.INFO)
        self.logger.disabled = False
        self.logger.propagate = True
        configure_logging(self.stream)

    def tearDown(self) -> None:
        for logger in (self.logger, logging.getLogger()):
            for handler in list(logger.handlers):
                logger.removeHandler(handler)
                handler.close()
        self.stream.close()

    def test_single_event_has_one_record(self) -> None:
        self.logger.info("订单校验完成", extra={"event": "order_validated"})
        lines = self.stream.getvalue().splitlines()
        self.assertEqual(len(lines), 1)
        record = json.loads(lines[0])
        self.assertEqual(record["event"], "order_validated")
        self.assertEqual(record["message"], "订单校验完成")

    def test_repeated_configuration_does_not_accumulate(self) -> None:
        configure_logging(self.stream)
        configure_logging(self.stream)
        self.logger.info("入口初始化完成")
        self.assertEqual(len(logging.getLogger().handlers), 1)
        self.assertEqual(len(self.stream.getvalue().splitlines()), 1)

    def test_handler_level_filters_debug(self) -> None:
        self.logger.setLevel(logging.DEBUG)
        self.logger.debug("调试校验完成")
        self.logger.info("业务校验完成")
        self.assertEqual(len(self.stream.getvalue().splitlines()), 1)

    def test_root_force_does_not_remove_child_handler(self) -> None:
        self.logger.addHandler(logging.StreamHandler(self.stream))
        configure_logging(self.stream)
        self.logger.info("重复路径复现")
        self.assertEqual(len(self.stream.getvalue().splitlines()), 2)

    def test_stopping_propagation_without_handler_loses_info(self) -> None:
        self.logger.propagate = False
        self.logger.info("无输出路径的事件")
        self.assertEqual(self.stream.getvalue(), "")

    def test_exception_record_keeps_error_context(self) -> None:
        try:
            raise ValueError("订单状态不合法")
        except ValueError:
            self.logger.exception("订单校验失败")
        lines = self.stream.getvalue().splitlines()
        self.assertEqual(len(lines), 1)
        record = json.loads(lines[0])
        self.assertIn("ValueError", record["exception"])
        self.assertEqual(record["level"], "ERROR")


if __name__ == "__main__":
    unittest.main()

执行：

python -m unittest -v test_logging_setup.py

应通过六项测试。第四项刻意保留错误路径，证明只给 root 加 force=True 仍会重复；第五项证明只关传播可能丢日志。第三项则确保 INFO 输出器不会因为子 logger 显式开启 DEBUG 而泄漏调试记录。

上线验收还要覆盖：冷启动输出一次、实际框架初始化后输出一次、原始 stderr 与日志平台数量一致、框架访问日志仍存在。测试中的重配置是回归验证，生产入口依然应在启动时配置一次，不在请求处理中修改全局日志配置。

六、预防与回滚

代码评审时把日志配置视为进程级资源管理：库模块使用命名 logger，必要时安装 NullHandler；输出路径由应用入口决定。Python Logging HOWTO

将初始化后的 handler 数量和归属作为启动验收内容。不要在每个类构造函数里“保证日志可用”，也不要以降低告警阈值或平台去重掩盖配置增长。

迁移先在单个实例验证。若出现业务日志缺失或框架访问日志消失，恢复上一份启动配置并重启该实例；保留本次传播路径快照，用它确认哪个输出归属被错误覆盖。不要在运行中继续叠加 handler 试图修复。

总结

先确认一条业务事件是否在同一进程的原始输出中重复，再查看 logger 到祖先的有效传播路径。修复后同时验证“只输出一次”和“需要的日志仍能输出”，才能避免把重复日志问题变成日志丢失问题。