多进程Python程序周期性出现MemoryError，求排查与处理方案

针对你遇到的多进程队列pickle阶段触发的MemoryError，我来拆解下可能的原因和可行的修复方案：

可能的原因 #

• 待序列化对象体积过大：当你往多进程队列推送的单个对象（比如超大字典、完整数据集）占用内存过高时，ForkingPickler.dumps()在将对象转换为字节流的过程中需要额外内存处理，一旦超过当前进程或系统剩余内存阈值，就会触发MemoryError。
• 队列数据积压严重：如果生产者推送数据的速度远快于消费者处理速度，队列会堆积大量未处理对象，内存占用持续攀升，最终耗尽可用内存。
• 循环引用引发内存泄漏：若你的对象存在互相引用的循环（如A引用B、B又引用A），pickle在序列化时可能无法正确释放临时内存，长期运行后内存占用逐渐累积，最终触发错误。
• Python 3.4版本固有问题：Python 3.4早已停止维护，multiprocessing模块的ForkingPickler或队列实现可能存在未修复的内存管理bug，比如特定场景下的内存泄漏。

捕获错误并让程序持续运行的方法 #

1. 在生产者端直接捕获异常 #

最直接的方式是在调用queue.put()的代码块外层添加异常捕获，这样单个对象序列化失败时，程序可以记录日志、跳过该对象，继续运行：

from multiprocessing import Queue
import logging
import time

logging.basicConfig(level=logging.ERROR)
queue = Queue()

def generate_data():
    # 模拟生成可能的大对象
    return {"large_data": [i for i in range(10**6)]}

def produce_data():
    while True:
        data = generate_data()
        try:
            queue.put(data)
        except MemoryError:
            logging.error("Failed to push data to queue: MemoryError occurred")
            # 可选：添加延迟或告警逻辑，避免瞬间重复报错
            time.sleep(1)
            continue

2. 子类化Queue重写_feed方法（进阶） #

如果异常是在队列内部的_feed方法中抛出，你可以子类化multiprocessing.Queue重写_feed方法，在内部捕获MemoryError，避免整个进程崩溃：

from multiprocessing import Queue
import logging

class SafeQueue(Queue):
    def _feed(self):
        try:
            super()._feed()
        except MemoryError:
            logging.error("MemoryError caught in queue feed process")
            # 可添加清理逻辑，比如通知主进程队列异常、重置状态等
            # 注意：多进程环境下需保证线程/进程安全

⚠️ 这种方式修改了队列内部逻辑，建议先在测试环境充分验证后再上线。

3. 优化序列化与队列管理 #

• 拆分大对象：不要一次性推送超大对象，拆分为多个小对象分批传输，降低单次序列化的内存压力。
• 替换序列化方式：用更高效的序列化库（如msgpack）替代默认的ForkingPickler，这类库通常内存占用更低、速度更快。可以在放入队列前先序列化为字节流，再传输：

import msgpack

# 生产者端
data = generate_data()
serialized_data = msgpack.packb(data)
queue.put(serialized_data)

# 消费者端
serialized_data = queue.get()
data = msgpack.unpackb(serialized_data)

• 监控队列长度：在生产者端检查队列qsize()，当队列长度超过阈值时暂停生产，等待消费者处理：

MAX_QUEUE_SIZE = 100
def produce_data():
    while True:
        if queue.qsize() < MAX_QUEUE_SIZE:
            data = generate_data()
            try:
                queue.put(data)
            except MemoryError:
                logging.error("MemoryError when pushing data")
        else:
            # 队列已满，短暂休眠后重试
            time.sleep(0.5)

4. 升级Python版本（推荐） #

Python 3.4早已停止维护，后续版本（如3.7+）修复了大量内存管理和多进程相关的bug。升级到稳定的新版本，大概率能解决这类底层内存问题。

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