为什么多进程场景下会获取到相同的logging logger实例并生成重复日志？

问题根因 #

你遇到的重复日志核心原因是进程池的工作进程会被复用，而Python的logging模块的logger实例是进程内全局单例的：

• 你向进程池提交了10次任务，但进程池默认的工作进程数和CPU核心数一致，部分工作进程会承接多个任务
• 同一个工作进程内第一次执行get_queue_logger时已经给root logger添加了QueueHandler，第二次执行同个函数时会再次添加Handler，同一条日志会被多个Handler处理输出，就出现了重复日志
• 你控制台输出里相同id的logger就是同一个工作进程内的实例，也验证了这个问题

解决方法 #

方法1：添加Handler前先做去重判断 #

最简单的修复方式是在添加Handler前，先判断root logger是否已经添加过同类型的Handler，避免重复添加：

import logging
from logging.handlers import QueueHandler
from multiprocessing import Manager, Pool

def get_queue_logger(q):
    root = logging.getLogger()
    # 先判断是否已经添加过QueueHandler，避免重复添加
    has_qh = any(isinstance(hdl, QueueHandler) for hdl in root.handlers)
    if not has_qh:
        qh = QueueHandler(q)
        print(f"Logger: {id(root)}, Handlers: {root.handlers}")
        root.addHandler(qh)
    return root

def main():
    q = Manager().Queue()
    pool = Pool()
    for i in range(10):
        pool.apply(
            func=get_queue_logger,
            args=(q,)
        )
    pool.close()
    pool.join()

if __name__ == "__main__":
    main()

方法2：使用标准多进程日志方案（推荐） #

多进程写同一个日志文件更稳妥的方案是用「单日志进程消费队列写文件」的架构，完全避免多进程写文件冲突和重复日志问题：

import logging
from logging.handlers import QueueHandler, QueueListener
from multiprocessing import Manager, Pool

# 日志监听进程逻辑，统一负责写日志到文件
def init_log_listener(q, log_path):
    file_handler = logging.FileHandler(log_path, encoding="utf-8")
    formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
    file_handler.setFormatter(formatter)
    listener = QueueListener(q, file_handler)
    listener.start()
    return listener

# 工作进程的业务逻辑，只负责把日志发到队列
def run_task(q, task_id):
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    # 仅第一次给当前进程的logger加QueueHandler
    if not any(isinstance(hdl, QueueHandler) for hdl in root.handlers):
        root.addHandler(QueueHandler(q))
    root.info(f"执行任务{task_id}")

def main():
    q = Manager().Queue()
    # 先启动日志监听进程
    log_listener = init_log_listener(q, "run.log")
    try:
        pool = Pool()
        for i in range(10):
            pool.apply(run_task, args=(q, i))
        pool.close()
        pool.join()
    finally:
        # 所有任务结束后再停止监听，保证所有日志落盘
        log_listener.stop()

if __name__ == "__main__":
    main()

内容的提问来源于stack exchange，提问作者lvcha