使用multiprocessing时内存缓存无法清理的问题求助

你使用multiprocessing.Pool优化计算，每个StarterClass实例加载约500MB数据，初始运行正常，但随着进程推进内存占用持续上涨，最终耗尽RAM。已在finally块尝试清理，但效果不佳，且results内存占比极低，问题根源不在此。

核心代码片段 #

主进程启动代码：

if __name__ == "__main__":
    params = list(itertools.product(cutoff_hours, cutoff_minutes, chunk_sizes, buffer_sizes, products, optimize_ons))

    params_dilled = [dill.dumps(param) for param in params]

    with multiprocessing.Pool(processes=20, initializer=worker_init, maxtasksperchild=10) as pool:
        results = pool.map(compute_for_parameters, params_dilled)

Worker初始化函数：

def worker_init():
    globals()["dill"] = dill

任务处理函数：

def compute_for_parameters(params_dilled):
    try:
        params = dill.loads(params_dilled)

        var1, var2, var3 = params

        try:
            starter = StarterClass(
                var1=var1,
                var2=var2,
                var3=var3,
            )

            results = starter.get_data()

            return results
        except Exception as e:
            logging.error(f"Error processing parameters {params}: {e}")
            raise
    finally:
        logger.info(f"Clearing cache...")
        del pnl, results
        gc.collect()

解决方案 #

1. 强化进程内核心资源清理

你当前的清理逻辑遗漏了加载500MB数据的StarterClass实例starter，这才是内存占用的核心，必须显式删除：

finally:
    logger.info(f"Clearing cache...")
    # 显式删除StarterClass实例及相关变量
    del starter, pnl, results
    # 强制触发垃圾回收
    gc.collect()

如果StarterClass内部持有文件句柄、数据库连接等外部资源，需要在类中实现cleanup()方法，或者在__del__中主动释放这些资源，避免隐式引用残留。

2. 调小maxtasksperchild参数

你已设置maxtasksperchild=10，可以尝试将值调至1或2——让进程处理更少任务就被销毁重建，彻底清除进程内积累的内存碎片和未释放资源。虽然会增加进程重建的开销，但对于内存泄漏严重的场景，这个取舍是合理的。

3. 移除全局变量的隐式引用

worker_init中将dill放入全局变量可能导致隐式引用无法回收，直接在任务函数内导入即可：

def compute_for_parameters(params_dilled):
    import dill  # 函数内导入，避免全局引用残留
    try:
        params = dill.loads(params_dilled)
        # ... 后续逻辑

4. 替换进程池实现或手动管理进程

如果内存泄漏问题始终无法缓解，可以尝试：

• 改用concurrent.futures.ProcessPoolExecutor，其资源管理逻辑与multiprocessing.Pool略有差异，可能避免部分泄漏场景；
• 手动创建单任务进程：每次处理一个参数就新建进程，处理完成立即销毁，彻底杜绝内存积累（代价是进程创建销毁开销增加）：

if __name__ == "__main__":
    params = list(itertools.product(...))
    results = []
    # 用队列获取进程返回结果
    result_queue = multiprocessing.Queue()
    
    def worker_wrapper(param, queue):
        res = compute_for_parameters(dill.dumps(param))
        queue.put(res)
    
    for param in params:
        p = multiprocessing.Process(target=worker_wrapper, args=(param, result_queue))
        p.start()
        p.join()
        results.append(result_queue.get())

5. 排查StarterClass内部的内存泄漏

内存泄漏的根源大概率在StarterClass的get_data()方法中，可通过以下方式定位：

• 检查类内是否有全局/静态变量在持续累积数据；
• 排查第三方库（如pandas、numpy）的使用是否存在未释放的对象（比如DataFrame缓存、未关闭的文件对象）；
• 用tracemalloc在worker进程内监控内存变化，定位具体代码行：

def compute_for_parameters(params_dilled):
    import tracemalloc
    tracemalloc.start()
    
    try:
        # ... 原有代码
        snapshot = tracemalloc.take_snapshot()
        top_stats = snapshot.statistics('lineno')
        logger.info("[Top 10 memory usage]")
        for stat in top_stats[:10]:
            logger.info(stat)
    finally:
        tracemalloc.stop()
        # ... 清理代码

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