Python多进程环境下日志重复输出问题求助

问题背景 #

主进程创建文件日志处理器并注册到根日志器，启动6个子进程后，AuthenticationFactory类的日志每条重复输出42次（7个进程×6次重复），pinger.py日志也存在相同重复问题。

代码结构 #

xyz
- main.py
- authentication/
  - authentication_factory.py（包含工厂类）
- pinger.py

代码详情 #

main.py #

from xyz.pinger import Pinger
from xyz.authentication.authentication_factory import AuthenticationFactory
import logging
import os
import multiprocessing
from functools import partial

_log = logging.getLogger()

class no_daemon_process(multiprocessing.Process):
    def _get_daemon(self):
        return False
    def _set_daemon(self, value):
        pass
    daemon = property(_get_daemon, _set_daemon)

class pool_nodaemon(multiprocessing.pool.Pool):
    Process = no_daemon_process

def target(group, params):
    client = AuthenticationFactory.get_auth_client(version=2)
    client.set_client_auth(group)
    Pinger(group).run()

def main():
    log_file = os.path.join(os.environ['LOG_FILE_PATH'],'{}.log'.format(os.environ['SERVICE_NAME']))
    _log.setLevel(logging.INFO)
    handler = logging.handlers.TimedRotatingFileHandler(log_file, when='H', interval=6, backupCount=10)
    formatter = logging.Formatter("%(asctime)-15s %(name)s %(levelname)s: %(message)s")
    handler.setFormatter(formatter)
    _log.addHandler(handler)
    
    groups = getGroups()
    pool = pool_nodaemon(len(groups)) # 实际groups长度为6
    action = partial(target, params=None)
    pool.map(target, groups)

if __name__ == '__main__':
    main()  

authentication_factory.py #

import logging

_log = logging.getLogger(__name__)

class AuthenticationFactory():
    @staticmethod
    def get_auth_client(version=2):
        if version == 1:
            _log.debug("Instantiating v1 authenticator")
            return v1Authenticator()
        elif version == 2:
            _log.debug("Instantiating v2 authenticator")
            return v2Authenticator()
        else:
            _log.error("Bad version: {}, cannot instantiate authenticator".format(version))
            return False

pinger.py #

import logging

_log = logging.getLogger(__name__)

class Pinger():
    pass # 内容与问题无关

观察现象 #

• 在authentication_factory.py中设置_log.propagate=False后，日志不再写入文件，转而输出到屏幕，且屏幕日志无重复；
• pinger.py的日志重复问题，设置propagate=False后得到解决。

问题根源 #

多进程环境下，子进程会完整继承主进程的日志处理器。主进程启动子进程时，根日志器的文件处理器会被每个子进程复制，导致同一条日志消息被主进程+所有子进程的处理器重复写入文件。同时，日志默认的propagate=True特性会让子日志器的消息向上传递到根日志器，进一步加剧重复次数。

解决方案 #

方案1：使用QueueHandler实现集中式日志处理（官方推荐） #

通过队列让主进程统一处理所有子进程的日志，彻底避免处理器重复和文件写入冲突：

修改main.py代码如下：

from xyz.pinger import Pinger
from xyz.authentication.authentication_factory import AuthenticationFactory
import logging
import logging.handlers
import os
import multiprocessing
from functools import partial

def listener_configurer(log_file):
    # 主进程配置日志写入逻辑
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    handler = logging.handlers.TimedRotatingFileHandler(log_file, when='H', interval=6, backupCount=10)
    formatter = logging.Formatter("%(asctime)-15s %(name)s %(levelname)s: %(message)s")
    handler.setFormatter(formatter)
    root.addHandler(handler)

def listener_process(queue, configurer, log_file):
    # 独立监听进程，负责从队列取日志并写入文件
    configurer(log_file)
    while True:
        record = queue.get()
        if record is None:
            break
        logger = logging.getLogger(record.name)
        logger.handle(record)

def worker_configurer(queue):
    # 子进程仅配置QueueHandler，将日志发送到队列
    root = logging.getLogger()
    root.handlers = []  # 清空继承自主进程的处理器
    root.setLevel(logging.INFO)
    handler = logging.handlers.QueueHandler(queue)
    root.addHandler(handler)

def target(group, params, queue):
    worker_configurer(queue)
    client = AuthenticationFactory.get_auth_client(version=2)
    client.set_client_auth(group)
    Pinger(group).run()

def main():
    log_file = os.path.join(os.environ['LOG_FILE_PATH'],'{}.log'.format(os.environ['SERVICE_NAME']))
    
    # 创建日志队列和监听进程
    queue = multiprocessing.Queue(-1)
    listener = multiprocessing.Process(target=listener_process, args=(queue, listener_configurer, log_file))
    listener.start()
    
    groups = getGroups()
    pool = pool_nodaemon(len(groups))
    # 将队列传递给子进程
    action = partial(target, params=None, queue=queue)
    pool.map(action, groups)
    
    # 发送结束信号，关闭监听进程
    queue.put(None)
    listener.join()

if __name__ == '__main__':
    main()  

说明：子进程的日志会通过队列发送给主进程的监听进程，由主进程统一写入文件，既解决了重复问题，也避免了多进程同时写文件导致的内容混乱。

方案2：临时解决（不推荐用于生产） #

如果不想使用队列，可在子进程启动时清空继承的日志处理器，并关闭子日志器的传播特性：

1. 在authentication_factory.py和pinger.py中添加_log.propagate=False；
2. 在target函数开头清空子进程的根日志器处理器：

def target(group, params):
    root_log = logging.getLogger()
    root_log.handlers = []
    # 后续逻辑不变

缺点：多个子进程仍可能同时写入文件，导致日志内容混乱，仅适合临时测试场景。

补充说明 #

• 设置propagate=False能解决重复是因为关闭了日志向上传播，但同时导致子日志器的消息无法传递到根日志器的文件处理器，因此只能输出到默认的控制台；
• 生产环境优先使用QueueHandler方案，这是Python官方文档推荐的多进程日志处理方式。

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