Python多进程分治算法实现及进程内日志失效问题咨询

我太懂这种头疼了——好不容易用multiprocessing把分治算法的效率提上来，结果子进程里的日志完全没动静，排查问题连个线索都没有！

问题根源 #

Python的multiprocessing会生成全新的子进程，它们看似继承了主进程的日志配置，但两个常见坑会导致日志失效：

• 如果主进程直接把日志绑定到文件或控制台，子进程的日志输出会因为文件句柄竞争被覆盖、截断甚至完全丢失
• 要是你在启动子进程之后才配置主进程的日志，子进程根本没同步到这个配置，自然不会输出任何日志

靠谱解决方案：用日志队列统一处理 #

最稳妥的思路是让主进程负责日志输出，子进程只需要把日志消息发送到一个跨进程共享的队列里。具体步骤如下：

1. 创建共享日志队列：用multiprocessing.Manager()生成可跨进程访问的队列
2. 主进程配置QueueListener：负责从队列中取出日志消息，统一输出到文件/控制台
3. 子进程配置QueueHandler：把日志消息发送到共享队列，交给主进程处理

示例代码改造 #

假设你的分治算法核心逻辑是这样的，我给你加上正确的日志配置：

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

# 主进程初始化日志监听器
def setup_logging(queue):
    # 定义日志格式，带上进程名方便排查
    formatter = logging.Formatter('%(asctime)s - %(processName)s - %(levelname)s - %(message)s')
    
    # 输出到控制台的处理器（也可以换成FileHandler输出到文件）
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    
    # 启动队列监听器，负责消费子进程发来的日志消息
    listener = QueueListener(queue, console_handler)
    listener.start()
    return listener

# 子进程执行的分片处理函数
def process_chunk(func, chunk, queue):
    # 子进程配置QueueHandler，所有日志都发往共享队列
    logger = logging.getLogger()
    logger.addHandler(QueueHandler(queue))
    logger.setLevel(logging.INFO)
    
    logger.info(f"开始处理分片，包含 {len(chunk)} 个元素")
    result = func(chunk)
    logger.info(f"分片处理完成，返回 {len(result)} 个结果")
    return result

# 你的分治算法主函数
def divide_and_conquer(func, input_list, num_processes=4):
    with Manager() as manager:
        queue = manager.Queue()
        listener = setup_logging(queue)
        
        # 拆分列表为多个分片
        chunk_size = max(1, len(input_list) // num_processes)
        chunks = [input_list[i:i+chunk_size] for i in range(0, len(input_list), chunk_size)]
        
        # 启动子进程并收集结果
        processes = []
        results = manager.list()
        
        def worker(chunk):
            res = process_chunk(func, chunk, queue)
            results.append(res)
        
        for chunk in chunks:
            p = Process(target=worker, args=(chunk,))
            processes.append(p)
            p.start()
        
        # 等待所有子进程完成
        for p in processes:
            p.join()
        
        # 停止日志监听器，释放资源
        listener.stop()
        
        # 合并所有分片结果
        merged_result = []
        for res in results:
            merged_result.extend(res)
        return merged_result

# 测试用例：对列表元素翻倍
def test_func(chunk):
    return [x*2 for x in chunk]

if __name__ == "__main__":
    input_data = list(range(10))
    final_result = divide_and_conquer(test_func, input_data)
    print("最终合并结果:", final_result)

关键注意事项 #

• 必须在if __name__ == "__main__"代码块里启动分治函数，这是Windows系统下multiprocessing的强制要求，也能避免Unix系统下的重复初始化问题
• 不要在子进程里直接使用主进程创建的logger实例，必须重新配置QueueHandler
• 记得在所有子进程完成后停止QueueListener，避免资源泄漏

这样改造后，子进程的日志就能正常输出，而且所有进程的日志都会有序地由主进程统一处理，不会出现混乱或丢失的情况。

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