Python multiprocessing队列规模增大后运行缓慢问题如何解决？

核心根因分析 #

• 生产消费流程完全串行：当前代码逻辑为等待所有生产者任务全部执行完成、所有数据写入队列后才启动消费逻辑，数据量较大时队列会堆积大量未处理数据，触发操作系统IPC管道缓冲区的限流机制，后续queue.put操作会自动阻塞，直接表现为整体运行速度大幅下降。
• multiprocessing.Queue 固有特性限制：普通multiprocessing.Queue基于操作系统管道实现，默认缓冲区大小有限，超出阈值后写入自动阻塞；而multiprocessing.Manager.Queue是由管理进程托管的跨进程实现，本身通信开销比普通Queue更高，换用后反而可能加剧性能损耗。
• 额外隐性开销：queue.empty()、queue.qsize()这类队列状态查询方法需要跨进程同步状态，频繁调用会产生额外性能开销，且多进程场景下返回结果是非实时准确的，容易引发逻辑问题。
• 代码显性bug：当前代码存在两处可直接触发报错的问题：ennumerate拼写错误，正确写法为enumerate；process函数定义仅接收1个参数，但调用时传入了i、item两个参数，会直接触发参数不匹配异常。

优化方案 #

1. 改为生产消费并行执行 #

单独启动独立的消费者进程，和生产者进程池同时运行，避免队列无限制堆积数据，参考实现如下：

import multiprocessing
from concurrent.futures import ProcessPoolExecutor
import os

def process(i, item):
    data = do_processing(item)
    return data

def produce(i, item, queue):
    data = process(i, item)
    queue.put(data)

def consume(queue):
    while True:
        data = queue.get()
        # 哨兵值判断生产流程结束
        if data is None:
            break
        save(data)

if __name__ == '__main__':
    # 设定队列最大长度，避免无限制堆积
    queue = multiprocessing.Queue(maxsize=os.cpu_count() * 2)
    records = load_records()
    # 提前启动消费者进程
    consumer_proc = multiprocessing.Process(target=consume, args=(queue,))
    consumer_proc.start()

    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
        print('produce items')
        for i, item in enumerate(records.items()):
            executor.submit(produce, i, item, queue)
    # 生产全部完成后传入哨兵值，通知消费者退出
    queue.put(None)
    consumer_proc.join()

2. 限制队列最大容量 #

初始化队列时传入maxsize参数，队列满时put操作会自动阻塞等待消费，避免队列无限制占用内存和IPC缓冲区，通常设置为工作进程数的1~2倍即可平衡生产消费速度。

3. 优化跨进程数据传输逻辑 #

如果单条业务数据体积较大，不要直接通过队列传输完整数据：可以先将数据写入本地临时文件/共享内存，队列仅传输文件路径/共享内存寻址标识，大幅降低序列化和跨进程传输的开销。

4. 避免依赖队列状态方法做流程控制 #

不要使用queue.empty()、queue.qsize()判断消费是否结束，改用哨兵值（如上方案中的None）通知消费终止，既减少不必要的跨进程同步开销，也能避免状态不准导致的逻辑错误。

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