Python多进程写入Manager字典为空问题的排查与解决

我需要将模型运行1000次并将其统计数据存储到字典中。由于模型本身可能未利用多进程，我计划使用Python的multiprocessing包。以下是我的代码：

import multiprocessing

class ensemble_model:
    def __init__(self, df, k) -> None:
        ## do some calculations
        self.r2_list = [...]
        self.rmse_list = [...]    

def worker(components, dic1, dic2):
    result = ensemble_model(
        df,
        k = components
    )
    dic1[components] = result.r2_list
    dic2[components] = result.rmse_list

r2_dic = multiprocessing.Manager().dict()
rmse_dic = multiprocessing.Manager().dict()

processes = []
for n_comp in range(1,1001):
    p = multiprocessing.Process(target=worker, args=(n_comp, r2_dic, rmse_dic))
    processes.append(p)
    p.start()

for p in processes:
    p.join()

问题在于最终得到的是空字典，而每个不同的components（n_comp）对应的列表应该存入字典中。我已尝试使用ChatGPT并查阅文档，但仍未找到问题所在。请问有人知道原因及解决方法吗？

核心原因 #

• 全局变量df无法被子进程访问：worker函数直接使用了df，但在多进程模式下（尤其是Windows系统），子进程不会继承主进程的全局变量，会触发NameError导致子进程直接退出，没有执行字典写入操作。
• 未捕获子进程异常：代码没有处理子进程中的错误，异常被静默忽略，无法定位问题根源。

具体解决步骤 #

1. 传递df作为worker函数参数 #

将df显式传入worker，确保子进程能获取到所需数据：

def worker(components, df, dic1, dic2):
    result = ensemble_model(
        df,
        k = components
    )
    dic1[components] = result.r2_list
    dic2[components] = result.rmse_list

启动进程时补充该参数：

p = multiprocessing.Process(target=worker, args=(n_comp, df, r2_dic, rmse_dic))

2. 添加异常捕获与日志 #

在worker中加入异常处理，方便排查问题：

import logging

logging.basicConfig(level=logging.INFO)

def worker(components, df, dic1, dic2):
    try:
        result = ensemble_model(df, k=components)
        dic1[components] = result.r2_list
        dic2[components] = result.rmse_list
        logging.info(f"完成components={components}的计算")
    except Exception as e:
        logging.error(f"components={components}计算失败: {str(e)}")

3. 限制并发进程数（推荐优化） #

直接启动1000个进程会耗尽系统资源，建议用multiprocessing.Pool控制并发数，效率更高：

from multiprocessing import Pool, Manager

def worker_wrapper(args):
    components, df, dic1, dic2 = args
    try:
        result = ensemble_model(df, k=components)
        dic1[components] = result.r2_list
        dic2[components] = result.rmse_list
        return components, "success"
    except Exception as e:
        return components, str(e)

if __name__ == "__main__":
    # 假设df已提前定义
    df = ... 
    manager = Manager()
    r2_dic = manager.dict()
    rmse_dic = manager.dict()
    
    # 并发数设为CPU核心数的2倍，可根据实际调整
    pool_size = multiprocessing.cpu_count() * 2
    with Pool(pool_size) as pool:
        args_list = [(n_comp, df, r2_dic, rmse_dic) for n_comp in range(1, 1001)]
        results = pool.map(worker_wrapper, args_list)
    
    # 检查失败任务
    for comp, res in results:
        if res != "success":
            print(f"任务{comp}失败: {res}")

4. 确认模型计算逻辑正确 #

确保ensemble_model的__init__方法中，r2_list和rmse_list是真实计算生成的，而非占位符[...]。

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