使用Python multiprocessing并行生成嵌套列表变慢如何解决？

问题产生原因 #

• 你调用了同步阻塞的pool.apply()方法：该接口每次提交任务后都会阻塞主进程，等待当前任务执行完成才会返回结果，相当于你虽然启动了进程池，但所有任务仍然是串行执行的，额外多出的进程启动、进程间数据序列化/反序列化开销直接导致总耗时高于串行版本。
• 任务提交过于频繁、粒度过小：你的代码在双层循环内每次仅提交单个f函数调用作为任务，累计要提交12000次任务，每次提交和返回结果都要走进程间通信流程，这类开销累加后也会抵消并行带来的收益。

正确实现方案 #

核心优化思路是用批量异步提交接口替代同步提交，同时增大单任务粒度，减少进程间通信的次数，参考实现代码如下：

import multiprocessing
import time

# 原模拟函数保持不变
def f(a, b):
    for i in range(100000):
        a = True
    return [a * b, a + b]

# 打包每个外层l的元素对应的所有计算逻辑，减少总任务数
def batch_process(x_val, G):
    return [f(x_val, g)[0] for g in G]

if __name__ == '__main__':
    G = [1, 2]
    l = [t for t in range(6000)]

    t1 = time.time()
    my_list1 = [[f(l[x], G[i])[0] for i in range(len(G))] for x in range(len(l))]
    t2 = time.time()
    print("Original: ", t2 - t1)

    pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
    t3 = time.time()
    # 批量提交所有任务，starmap会自动把参数元组展开传入batch_process
    my_list2 = pool.starmap(batch_process, [(x_val, G) for x_val in l])
    t4 = time.time()
    pool.close()
    pool.join()
    print("Parallelized: ", t4 - t3)
    # 可取消注释验证结果一致性
    # assert my_list1 == my_list2

如果你不想额外封装批量处理函数，也可以先把所有参数对展开，批量执行完f后再重组为嵌套列表，实现逻辑如下：

# 展开所有参数对，批量提交执行
args = [(l[x], G[i]) for x in range(len(l)) for i in range(len(G))]
all_res = pool.starmap(f, args)
# 重组为目标嵌套结构
my_list2 = [[all_res[x*len(G)+i][0] for i in range(len(G))] for x in range(len(l))]

改造后的并行版本耗时通常可以降到串行版本的1/N（N为你设备的CPU核心数），如果实际场景中f函数的计算耗时更长，并行收益会更明显。

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