Python 日志重复输出：排查 handler 重复添加与 propagate 传播

脚本只执行了一次 logger.info()，终端却打印两遍；初始化函数多调用几次，重复次数还在增加。本文面向初级到中级读者，用 Python 标准库复现两种原因，再验证每次业务事件只输出一条日志。

如果问题是日志迟迟不出现，可先看 Python 输出缓冲排查。本文处理的是同一事件被重复输出，不需要安装第三方包。

1. 方法选择：先找出重复经过的 handler

常见路径有两条：

初始化函数每次创建新的 handler，并挂到同一个 logger 上。
子 logger 自带 handler，同时把记录传播给根 logger 的 handler。

Python 官方文档说明，祖先和子 logger 同时挂处理器可能造成重复输出。先检查路径，再选择统一入口配置，或让某个独立 logger 自己负责输出。

2. 准备环境

需要 Python 3.9+，无第三方依赖。示例在 Linux、Python 3.11.4 下验证，shell 命令适用于 Bash 和 zsh。

python3 --version
mkdir duplicate-log-demo
cd duplicate-log-demo

将下面代码保存为 logging_demo.py。每次命令都启动新的 Python 进程，避免上一次实验的配置残留。

3. 完整实现：复现与修复

"""Reproduce and fix duplicate logs in a fresh Python process."""
import argparse
import logging

FORMAT = "%(levelname)s %(name)s %(message)s"


def add_console(logger):
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(FORMAT))
    logger.addHandler(handler)


def local_logger():
    logger = logging.getLogger("demo.task")
    logger.setLevel(logging.INFO)
    logger.propagate = False
    if not logger.handlers:
        add_console(logger)
    return logger


def inspect_chain(logger):
    while logger is not None:
        levels = [logging.getLevelName(h.level) for h in logger.handlers]
        print(f"{logger.name}: handlers={len(logger.handlers)} "
              f"handler_levels={levels} propagate={logger.propagate} "
              f"effective={logging.getLevelName(logger.getEffectiveLevel())}")
        if not logger.propagate:
            break
        logger = logger.parent


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("mode", choices=[
        "repeat", "propagate", "fixed-root", "fixed-local",
    ])
    parser.add_argument("--inspect", action="store_true")
    args = parser.parse_args()
    logger = logging.getLogger("demo.task")
    logger.setLevel(logging.INFO)

    if args.mode == "repeat":
        logger.propagate = False
        for _ in range(2):
            add_console(logging.getLogger("demo.task"))
    elif args.mode == "propagate":
        logging.basicConfig(level=logging.WARNING, format=FORMAT)
        add_console(logger)
    elif args.mode == "fixed-root":
        for _ in range(2):
            logging.basicConfig(level=logging.INFO, format=FORMAT)
    else:
        for _ in range(2):
            logger = local_logger()

    if args.inspect:
        inspect_chain(logger)
    logger.info("job=42 completed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

查看帮助并运行两种故障模式：

python3 logging_demo.py --help
python3 logging_demo.py repeat
python3 logging_demo.py propagate

两个故障命令各自输出两行：

INFO demo.task job=42 completed
INFO demo.task job=42 completed

repeat 每次创建新的 StreamHandler；同名 logger 会被复用，所以它最终有两个处理器。propagate 则让记录分别经过当前 logger 和根 logger 的处理器。

检查传播链：

python3 logging_demo.py repeat --inspect
python3 logging_demo.py propagate --inspect

前者显示 demo.task: handlers=2；后者显示 demo.task 和 root 各有一个 handler。诊断信息写入 stdout，日志写入 stderr；终端中两种流的显示先后顺序不应作为判断依据。

propagate 模式故意把根 logger 设成 WARNING，INFO 仍会输出两遍。这是因为传播给祖先的是记录，祖先 logger 的级别不会再次过滤它；需要限制输出目的地时，设置对应 handler 的级别。可结合日志级别 API 文档检查有效级别和处理器级别。

4. 两种修复：按应用的配置归属选择

方法一：应用入口统一配置

python3 logging_demo.py fixed-root --inspect

demo.task 没有自己的 handler，根 logger 有一个；日志只出现一条。在独立脚本入口调用一次 basicConfig()，业务模块只获取 logger 并写记录。代码故意调用两次以验证：根 logger 已有 handler 时，普通 basicConfig() 不再添加处理器，见官方说明。

迁移已有程序时，删除业务模块里重复创建 console handler 的初始化代码，再重启进程。本例不会替已经运行的服务自动移除旧 handler。

方法二：独立 logger 负责自己的输出

python3 logging_demo.py fixed-local --inspect

local_logger() 在当前 logger 没有直接挂载的 handler 时才创建，并设置 propagate = False。这适合应用自己拥有的独立日志通道；需要按大小保留文件时，可以接着学习 RotatingFileHandler 日志轮转。

这里检查 logger.handlers，它只包含当前对象直接挂载的处理器。hasHandlers() 也可能查到祖先的处理器，不应直接拿来判断当前 logger 是否已配置。API 文档给出了它的查找规则。

这个简单的“检查再添加”只在启动阶段的单线程中执行，不要放进并发工作循环。两种方法选择一种；可复用库通常只获取命名 logger，把输出配置交给应用，参见库日志配置指南。

5. 验证实际输出和退出码

把下面代码保存为 check_logging.py，在同一目录运行 python3 check_logging.py：

import subprocess
import sys

expected = "INFO demo.task job=42 completed"
for mode, count in [
    ("repeat", 2), ("propagate", 2),
    ("fixed-root", 1), ("fixed-local", 1),
]:
    result = subprocess.run(
        [sys.executable, "logging_demo.py", mode],
        capture_output=True, text=True, timeout=5, check=True,
    )
    assert result.stdout == "", (mode, result.stdout)
    assert result.stderr.splitlines() == [expected] * count, (
        mode, result.stderr,
    )
    print(f"PASS {mode}: {count} record(s)")

invalid = subprocess.run(
    [sys.executable, "logging_demo.py", "unknown"],
    capture_output=True, text=True, timeout=5,
)
assert invalid.returncode == 2
assert "invalid choice" in invalid.stderr
print("PASS invalid mode: exit=2")

预期输出：

PASS repeat: 2 record(s)
PASS propagate: 2 record(s)
PASS fixed-root: 1 record(s)
PASS fixed-local: 1 record(s)
PASS invalid mode: exit=2

验证会启动独立子进程，检查精确日志内容、条数和错误参数退出码，不会只检查“程序没报错”。默认 StreamHandler 写入 stderr，因此这里读取 result.stderr。

6. 常见问题与直接修复

现象	操作
初始化一次多一行	把 handler 配置移到启动入口；独立通道可使用本例的 `local_logger()`。
子 logger 和根 logger 都输出	使用方法一只在根 logger 输出，或方法二关闭独立通道的传播。
修改根级别后仍有 INFO	检查子 logger 的有效级别和祖先 handler 的级别；在目标 handler 上调用 `setLevel(logging.WARNING)`。
`basicConfig()` 后格式没变	根 logger 可能已有处理器；由拥有配置的应用入口调整，避免在库里调用 `force=True`。
设置 `propagate = False` 后 INFO 消失	当前 logger 必须拥有可接收 INFO 的 handler，并检查自身有效级别。
单进程实验正常，线上仍重复	临时在格式中加入 `%(process)d`、`%(name)s`，检查是否有多个进程执行了同一个任务。

force=True 会移除并关闭根 logger 原有的 handler，也不会清除子 logger 自带的 handler；不要把它当作通用去重开关。若进程号不同，先排查重复启动；若原始日志只有一条而查看平台有两条，检查采集路径。可按 rg 日志排障流程先对照原始文件，避免用消息文本去重掩盖真实的重复执行。

总结

先检查当前 logger 的 handler 数量和传播链，再确定由入口还是独立通道负责输出。修复后用独立进程验证精确条数，确认重复初始化不会重复添加处理器。多进程执行和日志采集重复需要继续查来源，不能仅靠关闭传播解决。