如何实现每个进程占用N核的Python multiprocessing.pool替代方案？

Python的multiprocessing.pool模块默认每个进程占用1核，有没有等价工具可以实现每个进程占用N核？也就是实现类似示例中multiprocessing.cool_new_pool的功能：

import multiprocessing
def func(x):
    return x**2

TOTAL_CORES = 10 # 实际场景可用os.cpu_count()
N_CORES_PER_PROCESS = 2
# p = multiprocessing.Pool(N_CORES) # 普通进程池用法
p = multiprocessing.cool_new_pool(TOTAL_CORES , N_CORES_PER_PROCESS) # 期望的功能
results = p.map(func, [i for i in range(10000)])

实现思路与方案 #

• 核心逻辑拆解：你要的效果本质是「进程内再做并行」——让每个工作进程内部利用N核资源，结合外层进程池实现总核数的高效利用。因为Python的GIL限制，CPU密集型任务需要通过嵌套进程/绕GIL的库来实现，IO密集型则可以用嵌套线程。

• 手动实现类似cool_new_pool的功能：

  1. 先计算外层进程池的大小：pool_size = TOTAL_CORES // N_CORES_PER_PROCESS，比如10核/2核每进程，就是开5个外层进程。
  2. 每个外层进程内部开子进程池/线程池，处理分配到的任务分片。
     示例代码：

  import multiprocessing
  from functools import partial
  
  def func(x):
      return x**2
  
  def process_batch(batch, n_cores):
      # 进程内开子进程池处理当前批次（CPU密集型用进程绕GIL）
      with multiprocessing.Pool(n_cores) as inner_pool:
          return inner_pool.map(func, batch)
  
  if __name__ == "__main__":
      TOTAL_CORES = 10
      N_CORES_PER_PROCESS = 2
      pool_size = TOTAL_CORES // N_CORES_PER_PROCESS
      
      # 拆分任务为多个批次，每个外层进程处理一个批次
      tasks = list(range(10000))
      batch_size = len(tasks) // pool_size
      batches = [tasks[i*batch_size : (i+1)*batch_size] for i in range(pool_size)]
      # 补全最后一个批次的剩余任务
      if len(tasks) % pool_size != 0:
          batches[-1].extend(tasks[pool_size*batch_size:])
      
      # 启动外层进程池，绑定每个进程的核数参数
      with multiprocessing.Pool(pool_size) as outer_pool:
          process_func = partial(process_batch, n_cores=N_CORES_PER_PROCESS)
          results = outer_pool.map(process_func, batches)
      
      # 合并所有结果
      final_results = [item for sublist in results for item in sublist]
  

• 现成工具替代：

  • 如果你不想手动写嵌套逻辑，joblib库的Parallel可以快速实现这种嵌套并行，它会自动处理进程/线程的调度：

    from joblib import Parallel, delayed
    
    TOTAL_CORES = 10
    N_CORES_PER_PROCESS = 2
    pool_size = TOTAL_CORES // N_CORES_PER_PROCESS
    
    def func(x):
        return x**2
    
    def process_batch(batch):
        # 内层用N核处理当前批次
        return Parallel(n_jobs=N_CORES_PER_PROCESS)(delayed(func)(x) for x in batch)
    
    tasks = list(range(10000))
    batches = [tasks[i*len(tasks)//pool_size : (i+1)*len(tasks)//pool_size] for i in range(pool_size)]
    # 外层用pool_size个进程，每个进程内层开N核
    final_results = Parallel(n_jobs=pool_size)(delayed(process_batch)(batch) for batch in batches)
    final_results = [item for sublist in final_results for item in sublist]
    

  • 额外提示：如果你的任务是基于numpy、pandas这类自带并行优化的库，直接设置环境变量（比如OMP_NUM_THREADS=N_CORES_PER_PROCESS）就能让每个进程自动占用N核，不需要额外嵌套并行。

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