Python多进程优化：通过累加结果节省内存

原代码的问题在于会一次性把所有进程的结果都存进内存再求和，这会导致内存直接被撑爆。利用multiprocessing.Pool.apply_async的callback参数，能在每个子进程跑完后立刻处理结果，实时累加到总数组里，处理完的单个结果会被自动回收，完全不用堆积内存。

实现代码（全局变量版） #

import multiprocessing as mp
import numpy as np

def simulate():
    # 这里替换成你的物理模拟逻辑
    return np.random.rand(20, 2500, 2500)  # 示例生成随机结果数组

def accumulate_result(pictures):
    # 回调函数：把单个进程的结果累加到总结果里
    global result
    result += pictures

if __name__ == '__main__':
    nsteps = 20
    nres = 2500
    iterations = 10  # 你的总模拟次数
    
    # 初始化总结果数组
    result = np.zeros([nsteps, nres, nres])
    
    pool = mp.Pool(6)
    for _ in range(iterations):
        # 每个任务完成后自动触发回调函数
        pool.apply_async(simulate, callback=accumulate_result)
    
    # 关闭进程池，不再接受新任务
    pool.close()
    # 等待所有子进程执行完毕
    pool.join()
    
    # 如果需要平均值，直接除以迭代次数
    # result /= iterations

更优雅的无全局变量版（面向对象） #

如果不想用全局变量，可以用一个简单的累加器类来封装结果：

import multiprocessing as mp
import numpy as np

def simulate():
    # 替换成你的模拟逻辑
    return np.random.rand(20, 2500, 2500)

class ResultAccumulator:
    def __init__(self, shape):
        self.total = np.zeros(shape)
    
    def add(self, pictures):
        self.total += pictures

if __name__ == '__main__':
    nsteps = 20
    nres = 2500
    iterations = 10
    
    accumulator = ResultAccumulator([nsteps, nres, nres])
    pool = mp.Pool(6)
    
    for _ in range(iterations):
        pool.apply_async(simulate, callback=accumulator.add)
    
    pool.close()
    pool.join()
    
    result = accumulator.total
    # result /= iterations

核心说明 #

• 回调函数是在主进程里执行的，子进程只负责把计算好的数组传回主进程，不用处理共享内存的问题，逻辑简单且安全。
• 每个子进程的结果在累加完成后会被Python垃圾回收机制自动清理，内存占用始终只保留总结果数组+单个进程结果的大小，完全避免了内存爆炸。
• 不用手动遍历进程对象调用get()，回调机制会自动处理完成的任务，代码简洁优雅。

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