如何配置Python多进程池以充分利用7个CPU核心？

问题根源分析 #

你的代码当前启用了7个进程，但CPU核心未跑满，主要问题集中在以下几点：

1. 任务粒度不合理+chunksize配置错误：self.segmentsContainer仅包含7-8个分段，每个分段作为单个大任务分配给进程。若某分段处理耗时远长于其他，会导致其他进程完成任务后闲置。此外你设置的chunksize=self.totalNumOfTasksSegments完全无效——pool.map的chunksize是用来拆分任务列表的块大小，你的任务列表只有7-8个元素，把chunksize设为每个分段的行数（远大于7），等于每个进程只处理一个任务块，完全没起到负载均衡的作用。
2. 进程数设置保守：8核机器无需刻意减1预留核心，Windows系统下直接使用cpu_count()即可。
3. 潜在IO瓶颈：如果run函数包含磁盘读写、网络请求等IO操作，进程会进入等待状态，CPU自然无法跑满。

针对性解决方案 #

1. 调整任务粒度与chunksize（核心优化）

如果分段内的行可以独立处理，建议将任务拆分为细粒度的行单元，让进程能持续处理小任务，避免闲置：

from multiprocessing import cpu_count

# 扁平化所有分段，生成细粒度任务列表
all_rows = []
for segment in self.segmentsContainer:
    all_rows.extend(segment)

# 直接使用全部8个核心
NUM_PROCESSES = cpu_count()
# 计算合理的chunksize：总任务数/进程数，避免块太小导致调度开销
chunksize = max(1, len(all_rows) // NUM_PROCESSES)

with Pool(processes=NUM_PROCESSES) as pool:
    # 修改run函数为处理单一行的run_single_row
    for res in pool.map(self.run_single_row, all_rows, chunksize=chunksize):
        mapResCnt += 1
        self.NDVIsPer10mX10mList.append(res[0])
        self.areasOfCoveragePerList.append(res[1])
        self.interceptionList.append(res[2])
        self.fourCornersInEPSG25832List.append(res[3])

如果必须以分段为单位处理，去掉错误的chunksize参数，让进程池自动分配任务：

NUM_PROCESSES = cpu_count()
with Pool(processes=NUM_PROCESSES) as pool:
    # 不指定chunksize，让map自动把7-8个任务分配到8个进程
    for res in pool.map(self.run, self.segmentsContainer):
        mapResCnt += 1
        self.NDVIsPer10mX10mList.append(res[0])
        self.areasOfCoveragePerList.append(res[1])
        self.interceptionList.append(res[2])
        self.fourCornersInEPSG25832List.append(res[3])

2. 排查并优化IO操作

如果run函数存在IO逻辑：

• 提前将所有需要读取的文件加载到内存，避免进程频繁等待磁盘IO；
• 写操作尽量批量执行，或用进程安全队列统一处理写入，减少IO等待时间。

3. 可选：改用imap_unordered提升效率

若任务处理顺序不影响结果收集，用imap_unordered代替map，进程完成任务后立刻返回结果，无需等待所有任务按顺序执行，能减少部分闲置时间：

with Pool(processes=NUM_PROCESSES) as pool:
    for res in pool.imap_unordered(self.run, self.segmentsContainer):
        # 后续结果收集逻辑不变
        mapResCnt += 1
        self.NDVIsPer10mX10mList.append(res[0])
        self.areasOfCoveragePerList.append(res[1])
        self.interceptionList.append(res[2])
        self.fourCornersInEPSG25832List.append(res[3])

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