如何在Python多进程中高效返回字典类型结果？

针对你当前方案的两个痛点，推荐使用**进程池（ProcessPoolExecutor或multiprocessing.Pool）**让子进程直接返回完整结果字典，彻底抛弃共享内存对象，具体实现如下：

1. 修改训练方法，返回完整结果字典 #

首先调整EqProp.train方法，让它直接返回包含所有需要参数的字典（支持嵌套结构），新增字段仅需在返回字典中添加即可：

def train(self, X, Y, batch_size, process_num, total_processes):
    # 原有训练逻辑执行...
    
    # 构造结果字典，支持任意嵌套结构
    return {
        "process_num": process_num,
        "layers": {"conv1": weights_conv1, "fc2": weights_fc2},  # 嵌套字典示例
        "loss": {"train": train_loss, "val": val_loss},
        "metrics": {"accuracy": acc_score}  # 新增字段直接在这里添加
    }

2. 使用进程池执行任务并收集结果 #

方案A：concurrent.futures.ProcessPoolExecutor（推荐，语法更简洁） #

from concurrent.futures import ProcessPoolExecutor

# 遍历你的output_nodes
for k, v in self.output_nodes.items():
    all_results = []
    
    # 创建进程池，指定并行进程数
    with ProcessPoolExecutor(max_workers=N_PROCESSES) as executor:
        # 提交所有训练任务
        futures = [
            executor.submit(
                EqProp(self).train,
                X, Y, self.batch_size, process_num, N_PROCESSES
            )
            for process_num in range(N_PROCESSES)
        ]
        
        # 逐个获取子进程返回的结果
        for future in futures:
            all_results.append(future.result())
    
    # 合并所有结果并保存到文件
    merged = {
        "layers": {},
        "loss": {},
        "metrics": {}
    }
    for res in all_results:
        proc_id = res["process_num"]
        merged["layers"][f"proc_{proc_id}"] = res["layers"]
        merged["loss"][f"proc_{proc_id}"] = res["loss"]
        merged["metrics"][f"proc_{proc_id}"] = res["metrics"]
    
    # 保存结果（用pickle支持非序列化对象，如numpy数组；用json则需先转成可序列化格式）
    import pickle
    with open(f"merged_results_{k}.pkl", "wb") as f:
        pickle.dump(merged, f)

方案B：multiprocessing.Pool #

如果更习惯原生multiprocessing模块，也可以用Pool实现：

import multiprocessing as mp

# 包装函数（适配Pool.map的单参数要求）
def train_task(args):
    self_inst, X, Y, batch_size, proc_num, total_procs = args
    return EqProp(self_inst).train(X, Y, batch_size, proc_num, total_procs)

for k, v in self.output_nodes.items():
    # 构造任务参数列表
    task_args = [
        (self, X, Y, self.batch_size, proc_num, N_PROCESSES)
        for proc_num in range(N_PROCESSES)
    ]
    
    # 启动进程池执行任务
    with mp.Pool(N_PROCESSES) as pool:
        all_results = pool.map(train_task, task_args)
    
    # 合并并保存结果（逻辑同方案A）
    merged = {
        "layers": {},
        "loss": {},
        "metrics": {}
    }
    for res in all_results:
        proc_id = res["process_num"]
        merged["layers"][f"proc_{proc_id}"] = res["layers"]
        merged["loss"][f"proc_{proc_id}"] = res["loss"]
        merged["metrics"][f"proc_{proc_id}"] = res["metrics"]
    
    import pickle
    with open(f"merged_results_{k}.pkl", "wb") as f:
        pickle.dump(merged, f)

方案优势 #

• 效率提升：完全移除共享内存对象和mutex锁，子进程计算完成后直接返回结果，避免锁竞争带来的性能损耗；
• 使用便捷：
  • 新增返回字段仅需修改train方法的返回字典，无需调整多进程相关代码；
  • 天然支持嵌套字典结构，只要返回的对象可序列化（用pickle可支持绝大多数Python对象）。

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