多线程/多进程无限循环处理消息引发MemoryError求助

问题背景 #

程序逻辑为无限循环等待消息，累积30秒的消息后拆分为chunk并行处理，先后尝试threading和multiprocessing.Pool实现，均在几次迭代后出现MemoryError，相关代码如下：

初始threading实现代码 #

def process_data(data, i):
    #process data for chunks
    parsed_records = []
for msg in msgs:
    #just deleting unnecessary keys and few key data manipulated
    parsed_records.append(record_data)

name=f"sample_{rnd}.json"
with open(name, "w") as outfile:
    outfile.write(parsed_records)
    return True

while True:
    threads = []
    for n in range(len(num_of_chunks)):
        t = threading.Thread(target=process_data, args=(num_of_chunks[n], n))
        threads.append(t)
        t.start()

    # Stop the threads
    for x in threads:
        x.join()

改用multiprocessing的代码 #

import multiprocessing
from faker import Faker
# Create Faker object to generate fake data for Producer
fake = Faker()

def myrandomdata(i,j):
    return fake.random_int(min = 1, max = j)

def divide_chunks(l, n):
    small_msgs = []
    for i in range(0, len(l), n):
        small_msgs.append(l[i:i + n])
    return small_msgs


def process_data(data, i):
    #process data for chunks
    parsed_records = []

    for msg in msgs:
        #just deleting unnecessary keys and few key data manipulated
        parsed_records.append(record_data)
    rnd = myrandomdata(1, 2000)
    name=f"sample_{rnd}.json"
    with open(name, "w") as outfile:
        outfile.write(parsed_records)
    return True

if __name__ == "__main__": 
    while True:
        #sample data
        msgs = [{"id":123,"min_position":7,...}]
        #msgs are nothing but bulk data recieving from server continuously am appending to msgs
        chunk_msgs = divide_chunks(msgs, 3)

        #clearing msgs to append next data after chunking previous data
        msgs.clear()
        with multiprocessing.Pool(len(chunk_msgs)) as pool:
            pool.starmap(process_data, [(chunk_msgs[n],n) for n in range(len(chunk_msgs))])

核心优化方案 #

1. 修复process_data的变量引用错误 #

当前代码中process_data遍历全局msgs而非传入的data参数，导致每个进程/线程重复处理全量数据，内存直接翻倍。修正后：

import json

def process_data(data, i):
    parsed_records = []
    # 遍历传入的chunk分片数据，而非全局msgs
    for msg in data:
        # 示例：删除冗余字段的处理逻辑
        record_data = {k: v for k, v in msg.items() if k in ["id", "min_position"]}
        parsed_records.append(record_data)
    
    # 注意：必须用json.dumps转成字符串再写入，直接写列表会报错
    rnd = myrandomdata(1, 2000)
    name = f"sample_{rnd}.json"
    with open(name, "w") as outfile:
        json.dump(parsed_records, outfile)
    return True

2. 限制进程池大小，避免过度创建进程 #

不要根据chunk数量动态创建进程池，固定使用CPU核心数的1-2倍即可，减少进程上下文切换和内存开销：

# 替换原有的进程池创建逻辑
pool_size = multiprocessing.cpu_count() * 2
with multiprocessing.Pool(pool_size) as pool:
    pool.starmap(process_data, [(chunk, idx) for idx, chunk in enumerate(chunk_msgs)])

3. 完善消息累积与内存清理逻辑 #

确保30秒累积逻辑正确实现，避免无限制堆积消息；同时在处理后立即释放内存：

import time

if __name__ == "__main__": 
    msgs = []
    last_flush_time = time.time()
    while True:
        # 模拟从服务器接收消息（替换为实际逻辑）
        new_msg = {"id": fake.random_int(), "min_position": fake.random_int()}
        msgs.append(new_msg)
        
        # 达到30秒累积阈值后触发处理
        if time.time() - last_flush_time >= 30:
            chunk_msgs = divide_chunks(msgs, 3)
            # 立即清空列表，释放内存
            msgs.clear()
            last_flush_time = time.time()
            
            # 并行处理分片
            pool_size = multiprocessing.cpu_count() * 2
            with multiprocessing.Pool(pool_size) as pool:
                pool.starmap(process_data, [(chunk, idx) for idx, chunk in enumerate(chunk_msgs)])

4. 复用进程池，避免频繁创建销毁 #

原代码每次循环重建进程池，会导致进程频繁启停的内存浪费，改为全局复用：

if __name__ == "__main__": 
    pool_size = multiprocessing.cpu_count() * 2
    # 全局创建一次进程池
    pool = multiprocessing.Pool(pool_size)
    msgs = []
    last_flush_time = time.time()
    
    try:
        while True:
            # 接收消息逻辑
            new_msg = {"id": fake.random_int(), "min_position": fake.random_int()}
            msgs.append(new_msg)
            
            if time.time() - last_flush_time >= 30:
                chunk_msgs = divide_chunks(msgs, 3)
                msgs.clear()
                last_flush_time = time.time()
                
                pool.starmap(process_data, [(chunk, idx) for idx, chunk in enumerate(chunk_msgs)])
    finally:
        # 程序退出时关闭进程池
        pool.close()
        pool.join()

5. 优化内存占用：逐行写入文件 #

避免在内存中存储过大的parsed_records列表，改用逐行写入方式降低内存压力：

def process_data(data, i):
    fake = Faker()
    rnd = fake.random_int(min=1, max=2000)
    name = f"sample_{rnd}.json"
    with open(name, "w") as outfile:
        outfile.write("[")
        first_record = True
        for msg in data:
            record_data = {k: v for k, v in msg.items() if k in ["id", "min_position"]}
            if not first_record:
                outfile.write(",")
            json.dump(record_data, outfile)
            first_record = False
        outfile.write("]")
    return True

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