Python3.x多进程单文件日志异常：仅主进程日志可写入

问题现象 #

实现了基于Thread的日志线程类Logger，以及基于Process的测试进程类。主进程的日志能够正常写入文件，但子进程的日志完全无法写入。该问题仅在Python3.x版本中出现，Python2.7环境下日志功能正常，运行平台为MacOS。

相关代码 #

File 1: main.py #

import sys
import time

from logger import log, Logger
from processes import Processes
from test_process import TestProcess
from test_process_2 import TestProcess2


def print_some_text(manager):
    while not manager.exit.is_set():
        log('Inside print_some_text')
        time.sleep(20)
    manager.wait_for_child_processes_to_exit()

def main():
    try:
        logger_thread = Logger("final.log", "test_logs")
        logger_thread.start()

        log("main(): starting")
        processes_manager = Processes(logger_thread)

        test_process_1 = TestProcess('process')
        processes_manager.add_process(test_process_1)
        test_process_1.start()

        test_process_2 = TestProcess2('process')
        processes_manager.add_process(test_process_2)
        test_process_2.start()

        print_some_text(processes_manager)
    except Exception as e:
        log("Error encountered")
        sys.exit(1)

if __name__ == '__main__':
    main()

File 2: processes.py #

import signal
from logger import log, stop_logging
from multiprocessing import Event

class Processes:
    def __init__(self, logger):
        self.processes = []
        self.exit = Event()
        self.logger = logger
        signal.signal(signal.SIGINT, self.exit_gracefully)
        signal.signal(signal.SIGTERM, self.exit_gracefully)

    def add_process(self, proc):
        self.processes.append(proc)

    def exit_gracefully(self, signum, frame):
        log(f"Shutting down processes. Signum: {signum}")
        for p in self.processes:
            p.terminate()
        self.exit.set()

    def wait_for_child_processes_to_exit(self):
        for p in self.processes:
            p.join()
        stop_logging()
        self.logger_thread.join()

File 3: test_process.py #

from multiprocessing import Process, Event

import time
from logger import log

class TestProcess(Process):
    def __init__(self, name):
        Process.__init__(self)
        self._name = name
        self.exit = Event()

    def run(self):
        try:
            while not self.exit.is_set():
                log("Inside test_process_1")
                time.sleep(10)
        except Exception as e:
            log("Error encountered in test_process_1")

    def terminate(self):
        log("terminate test_process_1")
        self.exit.set()

File 4: test_process_2.py #

from multiprocessing import Process, Event

import time
from logger import log

class TestProcess2(Process):
    def __init__(self, name):
        Process.__init__(self)
        self._name = name
        self.exit = Event()

    def run(self):
        try:
            while not self.exit.is_set():
                log("Inside test_process_2")
                time.sleep(10)
        except Exception as e:
            log("Error encountered in test_process_2")

    def terminate(self):
        log("terminate test_process_2")
        self.exit.set()

File 5: logger.py #

import logging
from logging.handlers import TimedRotatingFileHandler
from multiprocessing import Queue
from threading import Thread

LOG_QUEUE = Queue()

class Logger(Thread):
    def __init__(self, logfile, name):
        Thread.__init__(self)
        self._logger = logging.getLogger(name)
        self._logger.setLevel(logging.INFO)
        handler = TimedRotatingFileHandler(logfile, when="midnight", interval=1, backupCount=0)
        self._logger.addHandler(handler)
        self._syslog_open = False


    def run(self):
        while True:
            record = LOG_QUEUE.get()
            if record is None:
                break
            self._log(record)


    def _log(self, msg):
        self._logger.log(20, msg)

def log(msg):
    if msg is None:
        return
    LOG_QUEUE.put(msg)

def stop_logging():
    LOG_QUEUE.put(None)

问题原因 #

Python3在MacOS平台下，multiprocessing.Process默认采用spawn启动方式：该方式会重新启动一个Python解释器进程，并重新导入所有依赖模块，导致子进程中的全局变量LOG_QUEUE是全新的队列实例，和主进程中Logger线程监听的队列并非同一个，因此子进程的日志消息无法传递到主进程的日志写入线程。

而Python2默认采用fork启动方式，子进程会直接继承父进程的内存空间，包括同一个LOG_QUEUE实例，因此日志功能可以正常工作。

同时，processes.py中存在一处bug：wait_for_child_processes_to_exit方法中调用了未定义的self.logger_thread，实际应该使用初始化时传入的self.logger。

解决方案 #

方案1：强制使用fork启动方式（仅Unix/Linux/MacOS可用） #

在主进程入口处设置多进程启动方式为fork，确保子进程继承父进程的内存空间：

# 修改main.py的入口部分
if __name__ == '__main__':
    import multiprocessing
    multiprocessing.set_start_method('fork')
    main()

方案2：显式传递队列（跨平台兼容） #

避免依赖全局变量，将主进程的LOG_QUEUE显式传递给每个子进程，确保所有进程使用同一个队列：

1. 修改test_process.py和test_process_2.py，添加队列参数并实现独立的日志方法：

# test_process.py 修改后
from multiprocessing import Process, Event
import time

class TestProcess(Process):
    def __init__(self, name, log_queue):
        Process.__init__(self)
        self._name = name
        self.exit = Event()
        self.log_queue = log_queue

    def run(self):
        try:
            while not self.exit.is_set():
                self.log("Inside test_process_1")
                time.sleep(10)
        except Exception as e:
            self.log(f"Error encountered in test_process_1: {str(e)}")

    def log(self, msg):
        if msg is None:
            return
        self.log_queue.put(msg)

    def terminate(self):
        self.log("terminate test_process_1")
        self.exit.set()

test_process_2.py做相同修改。

2. 在main.py中传递队列给子进程：

# 修改main.py的相关部分
from logger import log, Logger, LOG_QUEUE

# ...

test_process_1 = TestProcess('process', LOG_QUEUE)
processes_manager.add_process(test_process_1)
test_process_1.start()

test_process_2 = TestProcess2('process', LOG_QUEUE)
processes_manager.add_process(test_process_2)
test_process_2.start()

3. 修复processes.py中的bug：

# 修改processes.py的wait_for_child_processes_to_exit方法
def wait_for_child_processes_to_exit(self):
    for p in self.processes:
        p.join()
    stop_logging()
    self.logger.join()  # 替换原有的self.logger_thread.join()

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