多进程优化失效问题排查及其他Python性能优化方案咨询

为什么你的多进程代码反而更慢？ #

你用了multiprocessing.Pool.apply()，这个方法是同步阻塞的——每次提交一个任务后，主进程会等待该任务执行完成，才会提交下一个任务。相当于你的4核池根本没并行起来，只是单进程跑，还额外加了进程间调度、数据传递的开销，自然耗时更长，CPU利用率上不去。

另外，函数里的print(num_elements)在多进程环境下会引发输出竞争，多个进程同时往stdout写数据，这也会带来额外的性能损耗。

修复多进程代码的正确写法 #

把apply()换成apply_async()异步提交任务，最后要调用pool.close()和pool.join()等待所有子进程完成后再计时：

import random
import multiprocessing
import time


def generate_and_sum_random_numbers(num_elements):
    total_sum = 0
    # 直接累加，避免创建列表的内存开销
    for _ in range(num_elements):
        total_sum += random.randint(0, 100)
    # 去掉print减少竞争开销，若需要记录可以用队列或文件
    # print(num_elements)


if __name__ == '__main__':
    start = time.time()
    CPU_core_number = 4
    pool = multiprocessing.Pool(processes=CPU_core_number)
    
    # 异步提交所有任务
    tasks = []
    for i in range(3000):
        task = pool.apply_async(generate_and_sum_random_numbers, args=(i,))
        tasks.append(task)
    
    # 等待所有任务完成
    for task in tasks:
        task.get()
    
    pool.close()
    pool.join()
    
    stop = time.time()
    print("Time needed:", stop - start)

除多进程外的其他优化方案 #

1. 优化函数本身的计算逻辑 #

• 避免不必要的内存分配：原代码里先创建random_numbers列表再求和，完全可以直接在循环里累加，省掉列表的创建、存储和销毁开销，像上面修复后的代码那样。
• 用更快的随机数生成器：Python标准库的random是纯Python实现的，速度较慢。可以改用numpy.random（C实现，批量生成更快），示例代码：

  import numpy as np
  def generate_and_sum_random_numbers(num_elements):
      return np.random.randint(0, 101, size=num_elements).sum()
  

2. JIT编译加速 #

用numba给函数加上JIT装饰器，把Python代码编译成机器码，CPU密集型任务速度能提升数倍甚至几十倍：

from numba import jit
import random

@jit(nopython=True)
def generate_and_sum_random_numbers(num_elements):
    total_sum = 0
    for _ in range(num_elements):
        total_sum += random.randint(0, 100)
    return total_sum

（注：numba对random的支持需要确保版本兼容，也可以用numba自带的随机数生成器）

3. 任务打包（减少进程调度开销） #

如果你的任务都是很小的计算单元，进程间的调度、数据传递开销会占比很高。可以把多个小任务打包成一个大任务，比如每100个小任务合并成一个，减少提交任务的次数：

def batch_task(numbers_list):
    total = 0
    for num in numbers_list:
        total += sum(random.randint(0,100) for _ in range(num))
    return total

# 主进程里打包任务
batch_size = 100
batches = [range(i, i+batch_size) for i in range(0, 3000, batch_size)]
tasks = [pool.apply_async(batch_task, args=(batch,)) for batch in batches]

4. 多线程？（不推荐CPU密集型） #

因为Python的GIL（全局解释器锁），多线程在CPU密集型任务下几乎无法并行，所以不推荐。但如果你的函数里有IO操作（比如读写文件、网络请求），多线程才会有优势。

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