Python multiprocessing多进程并行时如何正确聚合访问子进程返回结果
问题根因
multiprocessing.Manager生成的Namespace属于进程间代理对象,这类代理只识别自身顶层属性的替换操作,不会追踪嵌套可变对象的内部修改。
原代码给ns.results赋值的是主进程本地的原生Python列表,后续对这个列表执行的append操作,只修改了本地临时对象的内存,完全没有触发Manager层的跨进程同步,最终从共享命名空间读取ns.results时,自然只能拿到空值。
注意:
Pool.map的返回值本身就是所有子进程结果的主进程侧副本,绝大多数只需要主进程汇总结果的场景,完全不需要引入Manager共享存储,过度使用共享对象反而会引入不必要的报错和性能损耗。
正确实现方案
方案1:优先使用主进程本地列表聚合结果
这种方式无额外跨进程开销,也不存在共享对象同步误差,是日常并发聚合的首选实现。
直接在主进程定义普通列表,承接每个Pool执行完成后返回的结果即可:
import multiprocessing as mp import multi_process_functions as mpf # 主进程本地定义入参与结果存储列表 names = ['Bob', 'Joe', 'Brian'] aggregated_results = [] if __name__ == '__main__': # 执行第一组并发任务 with mp.Pool() as p: res_f2 = p.map(mpf.f2, names) aggregated_results.append(res_f2) print(res_f2) # 执行第二组并发任务 with mp.Pool() as p: res_f3 = p.map(mpf.f3, names) aggregated_results.append(res_f3) print(res_f3) # 输出完全符合预期的聚合结果 print(aggregated_results)
方案2:需要跨进程共享结果时,使用Manager托管的代理容器
如果存在子进程需要直接写入结果池的场景,不能在Namespace上绑定本地原生列表/字典,必须替换为Manager生成的带同步能力的代理容器(如manager.list()、manager.dict()),这类对象的所有修改都会被Manager层追踪并跨进程同步。
修正后的共享存储写法:
data_manager = mp.Manager() ns = data_manager.Namespace() ns.names = ['Bob', 'Joe', 'Brian'] # 用Manager生成的代理列表替代本地原生列表,保证修改可同步 ns.results = data_manager.list() if __name__ == '__main__': with mp.Pool() as p: res = p.map(mpf.f2, ns.names) ns.results.append(res) print(res) with mp.Pool() as x: res = x.map(mpf.f3, ns.names) ns.results.append(res) print(res) # 读取时可将代理列表转为普通列表展示 print(list(ns.results))
运行环境注意事项
针对Windows + Jupyter Notebook的运行环境,有两个必须遵守的规则:
- 所有多进程入口、Pool初始化操作必须包裹在
if __name__ == '__main__':代码块内,避免Windows无fork机制导致的无限递归生成子进程问题 - 被多进程调用的函数必须放在独立的可导入文件中,不能直接定义在Jupyter Notebook的单元格内,否则会触发子进程导入失败的报错
内容的提问来源于stack exchange,提问作者Branden-Pincince
相关产品推荐
相关产品推荐

