Python Logging多进程多线程安全写入日志文件实战

并发日志输出的核心思路

操作系统层面无法保证多个进程同时打开同一个文件句柄进行写入的原子性。Python 官方推荐的做法是：仅允许一个线程（通常放在主进程中）持有真正的文件写入器（如 RotatingFileHandler），其他工作进程将日志记录序列化为队列消息发送至 multiprocessing.Queue。主进程监听队列并交由标准 Handler 处理，从而彻底隔离并发风险。

第1步：构建安全的日志配置字典

使用 logging.config.dictConfig 集中管理格式、级别与输出目标。重点配置 RotatingFileHandler 实现按大小自动切割，防止单文件无限膨胀造成磁盘满载。

import logging
import logging.config
import logging.handlers
from multiprocessing import Process, Queue
import random
import time

# 日志配置字典
LOG_CONF = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'detailed': {
            'format': '%(asctime)s | %(processName)-10s(%(process)d) | %(levelname)-8s | %(message)s'
        }
    },
    'handlers': {
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'filename': '/log/app.log',
            'maxBytes': 5 * 1024 * 1024,
            'backupCount': 60,
            'encoding': 'utf-8',
            'formatter': 'detailed',
            'level': 'DEBUG'
        },
        'console': {
            'class': 'logging.StreamHandler',
            'formatter': 'detailed',
            'level': 'INFO'
        }
    },
    'loggers': {},
    'root': {
        'level': 'DEBUG',
        'handlers': ['console', 'file']
    }
}

第2步：主进程启动独立日志监听线程

在主进程中创建非阻塞后台线程，持续从 Queue 读取序列化后的 LogRecord 对象，并交给全局 Logger 处理。接收 None 作为终止信号。

def logger_listener(q):
    '''主进程日志监听函数'''
    while True:
        record = q.get()
        if record is None:
            break
        logger = logging.getLogger(record.name)
        logger.handle(record)

第3步：子进程注入队列处理器

子进程初始化时必须清理父进程继承的 Handler。否则在 fork 模式下会导致双份输出或文件锁冲突。清除后绑定 QueueHandler 即可。

def worker_process(name, q):
    # 关键：清空继承的Handlers，避免多进程环境双重输出
    root = logging.getLogger()
    root.handlers.clear()
    root.setLevel(logging.DEBUG)

    queue_handler = logging.handlers.QueueHandler(q)
    root.addHandler(queue_handler)

    loggers_list = ['worker_module.sub', 'app.task']
    levels = [logging.DEBUG, logging.INFO, logging.WARNING]
    for _ in range(50):
        lvl = random.choice(levels)
        logger = logging.getLogger(random.choice(loggers_list))
        logger.log(lvl, f'Task executed by {name}')
        time.sleep(0.01)

第4步：完整调度流程与优雅退出

按照先启动子进程、再启动监听线程的顺序执行。业务完成后逐一 join 子进程，向队列发送终止标记后等待监听线程收尾，确保所有缓冲日志落盘。

if __name__ == '__main__':
    # 1. 应用配置
    logging.config.dictConfig(LOG_CONF)
    
    # 2. 创建共享队列
    q = Queue(-1)
    
    # 3. 启动子进程
    workers = []
    for i in range(4):
        p = Process(target=worker_process, args=(f'Worker-{i+1}', q))
        workers.append(p)
        p.start()
    
    # 4. 启动监听线程
    listener = threading.Thread(target=logger_listener, args=(q,))
    listener.daemon = True
    listener.start()
    
    # 5. 等待工作进程结束
    for wp in workers:
        wp.join()
    
    # 6. 安全退出：通知监听线程停止并等待收尾
    q.put(None)
    listener.join()

跨平台关键注意事项

Windows 的 spawn 启动方式会重新执行模块顶层代码，而 Linux 通常使用 fork 继承父进程状态。若不在子进程入口显式调用 root.handlers.clear()，部分 Linux 环境下可能复制父进程的 Handler，导致重复输出或文件锁冲突。始终建议在工作进程内手动清理并重新注入 QueueHandler。此外，监控磁盘空间与定期验证 /log/app.log 的权限归属，可进一步保障生产环境的稳定性。

简要总结

通过 multiprocessing.Queue 中转日志记录，配合主进程单一监听线程与 RotatingFileHandler 轮转策略，能够彻底解决 Python 并发编程中的日志乱码与覆盖问题。掌握上述配置字典结构、队列消费逻辑与进程清理步骤，即可快速集成至各类微服务或批量任务架构中。 以上就是Python Logging多进程多线程安全写入日志文件的详细内容，更多关于Python并发日志处理的技术资料请关注本站其它相关文章！