Python多进程运行时如何动态分配剩余可用CPU核心数

DevOps 小满

2026-8-27

环境参数

设备搭载Windows 11系统，4任务多进程运行状态下读取到的CPU规格参数如下：

import psutil
import multiprocessing

print(multiprocessing.cpu_count())
# 12
print(psutil.cpu_count())
# 12
print(psutil.cpu_count(logical=False))
# 6
print(len(psutil.Process().cpu_affinity()))
# 12

业务背景

日常会在该设备上全天同时运行两个独立项目，两个项目运行过程中均会调用multiprocessing实现多进程计算。
此前每个项目处理的DataFrame最大行数为4，因此两个项目同时触发多进程任务时，总进程数最高为8，不会占满CPU资源，原有实现代码如下：

def stackoverflow(stack):
    df = create_dataframe(stack)
    if len(df) == 0:
        pass
    else:
        try:
            max_process = multiprocessing.cpu_count()-1 or 1
            pool = multiprocessing.Pool(max_process)
            list_pool = pool.map(data_event, zip(repeat(stack), df.iterrows()))
        finally:
            pool.close()
            pool.join()

现阶段计划取消DataFrame的行数限制，处理行数范围可从1行到最高1000行不等。如果为每个项目的Pool固定设置5个进程上限，当其中一个项目仅需处理2行数据、另一个需处理8行数据时，后者无法调用空闲CPU资源，会出现效率损耗。
此前曾尝试通过CSV文件交互写入当前占用进程数的方式实现动态分配（该示例仅作思路演示，存在竞态等诸多缺陷完全不可用）：两个项目分别读写对应标识的CSV文件，启动多进程前读取对方项目写入的已占用进程数，计算自身可用核心数写入自身标识文件，任务结束后将自身占用数重置为0，示例代码如下：

def stackoverflow(stack):
    df = create_dataframe(stack)
    if len(df) == 0:
        pass
    else:
        another_mp = int(list(csv.reader(open(r'C:\Users\Computer\Desktop\multiprocessing_B.csv')))[0][0])

        free_to_use = 10-another_mp

        with open(r'C:\Users\Computer\Desktop\multiprocessing_A.csv', 'w+', newline='', encoding='UTF-8') as f:
            f.write(free_to_use)

        try:
            max_process = free_to_use
            pool = multiprocessing.Pool(max_process)
            list_pool = pool.map(data_event, zip(repeat(stack), df.iterrows()))
        finally:
            pool.close()
            pool.join()
            with open(r'C:\Users\Computer\Desktop\multiprocessing_A.csv', 'w+', newline='', encoding='UTF-8') as f:
                f.write(0)

另一个项目仅需将代码中读写文件名的A、B标识互换即可实现对应逻辑，但该方案不具备生产可用性。

核心诉求：实现规范的动态进程分配逻辑，当系统中已有X个多进程运行时，新启动的多进程任务可自动使用总CPU核心数减去X后的剩余核心资源。

实现方案

优先选择无额外依赖、无竞态风险的系统负载感知方案，完全不需要在多个项目之间做自定义状态同步：

基于全局CPU负载动态计算可用进程数
借助psutil采样系统实时CPU使用率，结合总逻辑核心数计算当前可分配的进程数，同时设置合理的上下限，避免进程数为0或过度超配。该方案不限制同时运行的项目数量，后台运行的其他CPU密集型软件也会被纳入负载计算，自动避让资源占用。
可直接复用的实现代码如下：

import psutil
import multiprocessing
from itertools import repeat

def calc_available_process(min_proc=1, reserved_core=1):
    # 两次采样取平均，规避CPU使用率瞬时波动导致的计算偏差
    psutil.cpu_percent(interval=0.3)
    cpu_usage = psutil.cpu_percent(interval=0.3)
    total_core = multiprocessing.cpu_count()
    # 按空闲占比换算可用核心，预留固定核心给系统基础进程
    free_core = int(total_core * (1 - cpu_usage / 100)) - reserved_core
    # 钳制进程数范围，不低于最小值，不超过总核心数减预留值
    return max(min_proc, min(free_core, total_core - reserved_core))

def stackoverflow(stack):
    df = create_dataframe(stack)
    if len(df) == 0:
        return
    pool = None
    try:
        # 进程数不超过待处理任务行数，避免空进程浪费资源
        max_process = min(calc_available_process(), len(df))
        pool = multiprocessing.Pool(max_process)
        list_pool = pool.map(data_event, zip(repeat(stack), df.iterrows()))
        return list_pool
    finally:
        if pool:
            pool.close()
            pool.join()

注意事项

不需要严格要求进程数和空闲核心数1:1匹配，操作系统自带的进程调度器会自动在多个就绪进程之间分配CPU时间片，总进程数不超过逻辑核心数1.5倍时不会产生明显的调度开销。
禁止使用普通文本文件、注册表这类无原子性保证的介质做跨进程状态同步，这类实现天生存在竞态缺陷，并发触发任务时必然出现资源计算错误。
如果需要严格控制所有项目的总进程数上限，可以使用系统原生命名信号量做全局资源计数，相比文件同步方案原子性由操作系统保证，不会出现竞态问题，但会引入少量额外依赖，对绝大多数单机场景而言，上述负载感知方案已经足够稳定可用。

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