Python3多进程优化：每个进程需占用多线程场景的高效调度方案

问题核心原因 #

你当前实现的问题出在进程池的最大工作进程数配置上，和chunksize参数无关：

• 你初始化mp.Pool时传入了mp.cpu_count()，会创建20个工作进程，会同时启动12个分析任务（总样本数只有12，小于进程数）
• 每个任务对应外部pipeline会占用4个线程，总需求线程数达到48，远超过你20核的硬件上限，出现严重的CPU资源抢占，导致运行效率下降
• chunksize参数仅控制任务批量分配给工作进程的粒度，不会限制同时运行的最大任务数，无法解决你的问题

解决方案 #

直接将进程池的最大工作进程数设置为5即可，刚好匹配5个任务 * 4线程/任务 = 20核的硬件资源，不会出现资源抢占：

import multiprocessing as mp

# 也可以动态计算避免硬编码：max_workers = mp.cpu_count() // 4
max_workers = 5
with mp.Pool(max_workers) as pool:
    results = pool.starmap(my_wrapper, [(sample, other_arg) for sample in sample_list])

可选优化 #

如果你需要监控外部pipeline的运行状态，可以在my_wrapper中捕获subprocess的返回值，判断任务是否运行成功：

def my_wrapper(sample, other_arg):
    cmd = 'external_pipeline --threads 4 --sample {0} --other {1}'.format(sample, other_arg)
    proc = subprocess.Popen(shlex.split(cmd), stderr=subprocess.PIPE, stdout=subprocess.PIPE)
    stdout, stderr = proc.communicate()
    # 返回运行状态和错误信息方便排查
    return proc.returncode, stderr.decode("utf-8", errors="ignore")

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