Python多进程性能衰减严重，寻求优化方案及原因解析

问题描述 #

在使用Python multiprocessing 时遇到严重的性能衰减问题，编写了最小测试代码如下，分别在8核本地电脑和独占使用的80核服务器上测试，结果显示多进程的性能衰减（decay）非常严重；即使延长单线程任务时长，该问题仍未改善。请问如何优化该问题，这是否是Python多进程的正常表现？

测试代码：

import os
os.environ["OMP_NUM_THREADS"] = "1" # just in case the system uses multithrad somehow
os.environ["OPENBLAS_NUM_THREADS"] = "1" # just in case the system uses multithrad somehow
os.environ["MKL_NUM_THREADS"] = "1" # just in case the system uses multithrad somehow
os.environ["VECLIB_MAXIMUM_THREADS"] = "1" # just in case the system uses multithrad somehow
os.environ["NUMEXPR_NUM_THREADS"] = "1" # just in case the system uses multithrad somehow
import numpy as np
from datetime import datetime as dt
from multiprocessing import Pool
from pandas import DataFrame as DF

def trytrytryshare(times):        
    i = 0
    for j in range(times):
        i+=1
    return


def trymultishare(thread = 70 , times = 10):
    st = dt.now()
    args_l = [(times,) for i in range(thread)]               
    print(st)
    p = Pool(thread)
    for i in range(len(args_l)):
       p.apply_async(func = trytrytryshare, args = (args_l[i]))
    p.close()
    p.join()
    timecost = (dt.now()-st).total_seconds()
    print('%d threads finished in %f secs' %(thread,timecost))
    return timecost
   

if __name__ == '__main__':
    res = DF(columns = ['thread','timecost'])
    n = 0
    for j in range(5):
        for i in range(1,8,3):
            timecost = trymultishare(thread = i,times = int(1e8)) 
            res.loc[n] = [i,timecost]
            n+=1
        timecost = trymultishare(thread = 70,times = int(1e8)) 
        res.loc[n] = [70,timecost]
        n+=1
    res_sum = res.groupby('thread').mean()
    res_sum['decay'] =  res_sum.loc[1,'timecost'] / res_sum['timecost'] 

问题分析与优化方案 #

1. 这不是Python多进程的正常表现 #

你的测试任务是纯CPU计算密集型（简单循环累加），理论上在CPU核心充足的情况下，多进程应该接近线性加速比。出现严重衰减说明测试代码或运行环境存在额外开销。

2. 核心问题排查 #

• 进程创建与销毁开销：每次调用trymultishare都创建一个新的Pool，而不是复用。创建70个进程的开销在80核服务器上可能被放大，尤其是当任务本身的CPU耗时没有远超过进程启动开销时。
• 任务粒度与进程数量不匹配：在8核机器上测试70个进程，会导致大量进程上下文切换，严重消耗CPU资源；即使在80核服务器，70个进程同时运行，若任务是纯计算，理论上应该接近线性，但你的测试结果衰减严重，可能和进程池的初始化、任务分发的额外开销有关。
• apply_async的循环调用效率低：循环调用apply_async逐个提交任务，不如使用map或imap批量提交，减少任务分发的 overhead。

3. 具体优化措施 #

• 复用进程池：不要每次测试都创建新的Pool，可以在主函数中初始化一次进程池，或者在测试循环内复用。
• 调整进程数量：进程数量不应超过CPU核心数（逻辑核心），8核机器最多设为8，80核服务器最多设为80，过多的进程会导致上下文切换开销剧增。
• 批量提交任务：用pool.map替代循环apply_async，减少任务提交的额外开销：
  修改后的trymultishare函数示例：

  def trymultishare(thread = 8 , times = 10):
      st = dt.now()
      args_l = [times for _ in range(thread)]  # map接受单个参数的迭代器，无需包装元组
      print(st)
      with Pool(thread) as p:
          p.map(trytrytryshare, args_l)
      timecost = (dt.now()-st).total_seconds()
      print('%d threads finished in %f secs' %(thread,timecost))
      return timecost
  

• 优化任务粒度：确保单个任务的耗时足够长，远超过进程启动和任务分发的开销。比如将times设为更大的值（如1e9），减少相对开销占比。
• 检查系统调度：即使是独占服务器，也要确认是否存在其他进程抢占CPU资源，或者系统的CPU调度策略是否影响进程运行。

4. 验证优化效果 #

优化后重新测试，你会发现：

• 当进程数量等于CPU核心数时，加速比接近线性；
• 超过核心数后，性能会开始下降，但不会出现严重衰减；
• 复用进程池和批量提交任务能显著降低额外开销。

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