如何使用Python multiprocessing多次调用函数并将结果存入numpy数组

你原有写法的错误原因：

• Pool.map()要求第二个参数必须是可迭代对象，你直接传入单个整数i不符合接口要求
• 循环内反复调用map每次仅执行1次任务，完全没有利用多进程的并行优势，性能反而不如串行执行

方案1：适配map接口（推荐） #

仅需要给原有无参函数加一个占位参数，不需要修改原有逻辑，代码如下：

import numpy as np
from multiprocessing import Pool

# 原有函数新增一个占位参数，适配map的传参要求
def f(_):
    # 原有随机计算逻辑保持不变
    a = np.random.rand()
    b = np.random.rand()
    return a, b

if __name__ == '__main__':
    N = 1000  # 替换为你的实际试验次数
    process_num = 15  # 建议设置为CPU物理核心数，不要过高
    
    with Pool(process_num) as p:
        # 传入长度为N的可迭代对象，map会自动分配任务到多进程执行
        result_list = p.map(f, range(N))
    
    # 直接将结果列表转为numpy数组，形状自动为(N,2)，顺序和串行执行完全一致
    results = np.array(result_list)

方案2：不修改原有无参函数 #

如果不想改动原有f()的定义，可以用apply_async批量提交异步任务：

import numpy as np
from multiprocessing import Pool

# 原有无参函数不需要任何修改
def f():
    a = np.random.rand()
    b = np.random.rand()
    return a, b

if __name__ == '__main__':
    N = 1000
    process_num = 15
    
    with Pool(process_num) as p:
        # 批量提交N个异步任务
        tasks = [p.apply_async(f) for _ in range(N)]
        # 按提交顺序收集结果，顺序和串行执行一致
        result_list = [task.get() for task in tasks]
    
    results = np.array(result_list)

注意事项 #

• Windows系统下必须把多进程执行逻辑全部包裹在if __name__ == '__main__'块内，否则会出现进程反复启动的报错
• 如果函数内部用到随机数生成，建议在f()内部每次执行时单独设置随机种子，避免多进程继承父进程随机状态导致生成重复的随机结果
• 进程数不要超过CPU物理核心数太多，否则上下文切换的开销会抵消多进程的性能收益

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