Python 多进程日志互相交错时怎么设计输出

多个 Python 进程同时把日志写到终端或同一个文件时，最稳妥的做法不是给每个进程再加一把锁，而是让 worker 只提交 LogRecord，由主进程的 QueueListener 统一写出。这样可以避免一条日志还没写完就被另一条插入，也把文件句柄和格式化规则集中到一个地方。

先分清：启动方式和输出方式是两件事

spawn 会启动新的 Python 解释器并重新导入可导入的主模块，目标函数和传入参数需要能够被序列化；fork 则从父进程复制出子进程，容易让人误以为父进程已经配置好的 logger 就能安全共用。实际上，文件 handler、缓冲区和输出流的生命周期仍然需要明确管理。

如果每个 worker 都直接挂 StreamHandler 或 FileHandler，它们拥有多个写入口。终端上可能出现半行交错；文件则可能遇到格式重复、缓冲刷新时机不同，甚至多个进程同时维护同一个文件 handler。先把写出口收敛，才能谈启动方式。

把日志写入口收敛成一个队列出口

下面的示例显式使用 spawn 上下文，方便在 Windows、macOS 和 Linux 上采用同一套入口约束。worker 不创建文件 handler，只保留一个 QueueHandler；主进程负责格式化并写到终端。若要落文件，只需在 listener 后面再放一个文件 handler。

import logging
import logging.handlers
import multiprocessing as mp
import time


def configure_worker_logging(log_queue):
    # 子进程只负责提交记录，避免多个进程直接竞争输出流。
    root = logging.getLogger()
    root.handlers.clear()
    root.setLevel(logging.INFO)
    root.addHandler(logging.handlers.QueueHandler(log_queue))


def worker(task_id, log_queue):
    configure_worker_logging(log_queue)
    logger = logging.getLogger(__name__)
    for part in range(3):
        # 进程名和任务号用于恢复不同 worker 的业务上下文。
        logger.info("task=%s part=%s finished", task_id, part)
        time.sleep(0.02)


def main():
    # 显式选择上下文，Queue、Process 和 logger 的生命周期保持一致。
    ctx = mp.get_context("spawn")
    log_queue = ctx.Queue()

    stream = logging.StreamHandler()
    stream.setLevel(logging.INFO)
    stream.setFormatter(logging.Formatter(
        "%(asctime)s %(processName)s %(levelname)s %(message)s"
    ))
    listener = logging.handlers.QueueListener(
        log_queue, stream, respect_handler_level=True
    )
    listener.start()

    processes = [
        ctx.Process(target=worker, args=(task_id, log_queue), name=f"worker-{task_id}")
        for task_id in range(3)
    ]
    for process in processes:
        process.start()
    for process in processes:
        # 先等 worker 结束，再停止 listener，避免尾部记录尚未取走。
        process.join()

    listener.stop()
    failed = [p.name for p in processes if p.exitcode != 0]
    if failed:
        raise RuntimeError(f"worker failed: {failed}")


if __name__ == "__main__":
    # spawn 会重新导入主模块，进程创建必须放在安全入口内。
    main()

这个结构的关键不是让日志“排队后天然有全局顺序”，而是让每条记录拥有完整的入队边界。格式串中的 processName、任务号和分片号会保留定位信息；若业务需要严格排序，还要在记录中加入单调递增的事件序号，不能只依赖时间戳。

spawn、fork 和队列生命周期怎么检查

fork 不是“无需配置”的捷径：它继承父进程状态，可能把不应跨进程复用的资源一并带入子进程。spawn 的约束更明显，但也更容易暴露不可导入的目标函数、不可序列化参数和顶层副作用。无论采用哪一种，日志架构都可以保持为“worker 产生日志，主进程集中写出”。

回归时别只看终端是否整齐

先让每个 worker 输出固定数量的记录，再核对每个任务号是否完整出现、每个进程是否正常退出、日志是否仍带有进程名。把任务号写进消息或额外字段，比依赖日志时间更可靠。还要专门测试 worker 抛异常、队列为空和最后一条日志紧贴进程退出的情况。

如果只要求一条记录不被拆成两半，集中 listener 已经足够；如果要求“任务 0 的最后一条一定早于任务 1 的第一条”，则需要业务层序号、汇总排序或完成事件，不能把 multiprocessing.Queue 当成全局排序器。

常见问题

为什么不直接给 FileHandler 加 multiprocessing.Lock？

锁可以序列化部分写入，但每个进程仍然拥有各自的 handler、缓冲和异常路径，关闭与轮转也更复杂。集中写出口通常更易维护。

QueueHandler 能不能配 queue.SimpleQueue？

多进程场景优先使用 multiprocessing.Queue。它与当前进程上下文一致，也符合 Python 文档对 QueueListener 配合 multiprocessing 的建议。

用了队列还会丢日志吗？

进程被强制终止、队列写入失败或 listener 过早停止时仍可能丢记录。正常收尾应先等待 worker，再停止 listener；生产环境还应记录队列异常并设计降级策略。