Python使用Multiprocessing并行for循环填充Numpy数组效率低于串行的问题

此前查阅过不少相关技术帖子，但现有方案均不适配当前场景：尝试基于Python多进程实现for循环填充Numpy数组的并行化，测试代码如下：

import numpy as np
from multiprocessing import Pool
import time


def func1(x, y=1):
    return x**2 + y

def func2(n, parallel=False):
    my_array = np.zeros((n))
    
    # 并行版本逻辑
    if parallel:
        pool = Pool(processes=6)
        for idx, val in enumerate(range(1, n+1)):
            result = pool.apply_async(func1, [val])
            my_array[idx] = result.get()
        pool.close()

    # 串行版本逻辑
    else:
        for i in range(1, n+1):
            my_array[i-1] = func1(i)

    return my_array

def main():
    start = time.time()
    my_array = func2(60000)
    end = time.time()
    
    print(my_array)
    print("Normal time: {}\n".format(end-start))

    start_parallel = time.time()
    my_array_parallelized = func2(60000, parallel=True)
    end_parallel = time.time()

    print(my_array_parallelized)
    print("Time based on multiprocessing: {}".format(end_parallel-start_parallel))


if __name__ == '__main__':
    main()

代码逻辑可正常运行、输出结果正确，但多进程版本耗时远高于原生串行版本，实际运行输出如下：

[2.00000e+00 5.00000e+00 1.00000e+01 ... 3.59976e+09 3.59988e+09
 3.60000e+09]
Normal time: 0.01605963706970215

[2.00000e+00 5.00000e+00 1.00000e+01 ... 3.59976e+09 3.59988e+09
 3.60000e+09]
Time based on multiprocessing: 2.8775112628936768

核心疑问：是否存在更优的异步结果获取方式？当前实现存在什么问题？该需求最高效的实现方式是什么？

• 任务粒度过细，多进程开销远高于计算收益：func1单轮计算仅为一次平方加1的算术操作，6万次串行计算总耗时仅16ms，单轮计算耗时不足1微秒；而多进程场景下，单次任务分发、进程间通信、结果回传的开销就达到毫秒级，细粒度任务使用多进程天然会比单进程慢几个数量级。
• 阻塞式拉取结果完全消解了并行优势：当前代码每提交一个apply_async任务就立刻调用result.get()阻塞等待结果返回，相当于把并行任务强行改成「提交一个、等待返回一个」的串行模式，进程池的6个工作进程根本没有被充分调度，资源完全闲置。
• 高频小数据序列化放大了额外开销：多进程跨进程传递参数和结果需要经过pickle序列化，每次仅传递单个数值、回传单个数值的操作，累积下来的序列化耗时已经远超过计算本身的耗时。

• 优先匹配计算场景选择实现方式：如果是无依赖的数组逐元素运算，优先使用Numpy内置向量化运算，不需要引入多进程，性能会比纯Python for循环高几个数量级。
• 如果确实需要使用多进程（比如单轮计算逻辑复杂、无法向量化），需要按照规则优化调用方式：
  • 把细碎的单元素计算打包成块批量提交，大幅减少任务分发、序列化的次数
  • 使用进程池自带的批量映射接口替代逐任务提交、逐次等待的写法，让进程池可以自动调度任务并行执行
  • 配合合理的chunksize参数调整分块大小，平衡调度开销和并行度

优化后的并行实现参考：

import numpy as np
from multiprocessing import Pool
import time

def func1(x, y=1):
    return x**2 + y

def func2(n, parallel=False):
    if parallel:
        x_seq = np.arange(1, n+1)
        with Pool() as pool:
            # 批量提交任务，设置分块大小为1000，减少调度次数
            my_array = pool.map(func1, x_seq, chunksize=1000)
        my_array = np.array(my_array)
    else:
        # 串行版本直接用向量化运算，性能远高于Python for循环
        x_seq = np.arange(1, n+1)
        my_array = func1(x_seq)
    return my_array

注意：多进程并行的收益只有在单任务计算耗时足够长（通常单轮计算超过10微秒）的场景下才会体现，对于本次测试的极轻量算术运算，哪怕优化了多进程写法，耗时也不会低于原生Numpy向量化的单进程实现。

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