Python多线程/多进程池如何限制CPU核心用量且不固定核心亲和性

这个需求可实现，核心逻辑是利用进程池/线程池的初始化钩子，在每个子实例启动时动态设置CPU亲和性，无需提前手动指定固定核心ID，完全交由操作系统调度器自主选择最优核心分配、动态调整。

前置依赖 #

推荐使用跨平台的psutil库完成核心亲和性设置，比系统原生API兼容性更好，可覆盖Windows、Linux、macOS全平台：

pip install psutil

核心说明 #

1. 进程池适配性最好：进程拥有独立的资源上下文，每个子进程可以单独设置核心使用上限，完全匹配需求。
2. 线程池使用限制：同一进程下的所有线程共享CPU亲和性配置，因此线程池的亲和性设置会作用于整个进程的所有线程，无法给单个线程单独设置核心上限，若需要每个子实例单独限制，优先选择进程池。
3. 避免固定绑定弊端：我们不需要提前给子进程分配固定的核心ID，每次子进程启动时会自动选择当前负载最低的X个核心作为允许使用的范围，操作系统调度器仍可在该范围内动态调整进程的运行核心，不会出现绑定到弱性能核心、跨Chiplet/跨CPU调度的性能问题。

代码示例 #

1. multiprocessing.Pool 实现示例 #

以16核机器、创建4进程池、每个进程最多使用2个核心为例：

import multiprocessing
import psutil

def worker_init(max_cores_per_worker: int):
    current_process = psutil.Process()
    total_cores = psutil.cpu_count(logical=True)
    # 边界处理，避免设置的核心数超过系统总可用核心
    use_core_count = min(max_cores_per_worker, total_cores)
    # 获取所有核心的当前使用率，选择使用率最低的use_core_count个核心
    core_usage = psutil.cpu_percent(percpu=True)
    # 按使用率从小到大排序，取前use_core_count个核心的ID
    sorted_cores = sorted(enumerate(core_usage), key=lambda x: x[1])
    selected_core_ids = [core_id for core_id, _ in sorted_cores[:use_core_count]]
    # 设置当前进程的CPU亲和性为选中的核心，操作系统仍可在这些核心内动态调度
    current_process.cpu_affinity(selected_core_ids)

def demo_task(x):
    # 你的业务逻辑
    return x * x

if __name__ == "__main__":
    # 创建4进程池，每个进程最多使用2个核心
    pool = multiprocessing.Pool(
        processes=4,
        initializer=worker_init,
        initargs=(2,)
    )
    res = pool.map(demo_task, range(10))
    print(res)
    pool.close()
    pool.join()

2. concurrent.futures.ProcessPoolExecutor 实现示例 #

from concurrent.futures import ProcessPoolExecutor
import psutil

def worker_init(max_cores_per_worker: int):
    current_process = psutil.Process()
    total_cores = psutil.cpu_count(logical=True)
    use_core_count = min(max_cores_per_worker, total_cores)
    core_usage = psutil.cpu_percent(percpu=True)
    sorted_cores = sorted(enumerate(core_usage), key=lambda x: x[1])
    selected_core_ids = [core_id for core_id, _ in sorted_cores[:use_core_count]]
    current_process.cpu_affinity(selected_core_ids)

def demo_task(x):
    return x * x

if __name__ == "__main__":
    # 初始化钩子设置方式和multiprocessing.Pool一致
    with ProcessPoolExecutor(
        max_workers=4,
        initializer=worker_init,
        initargs=(2,)
    ) as executor:
        res = list(executor.map(demo_task, range(10)))
    print(res)

注意事项 #

• 若使用Unix/Linux系统且不想安装第三方依赖，可将psutil的亲和性设置替换为原生os.sched_setaffinity接口，逻辑完全一致。
• CPU亲和性仅限制进程可使用的核心范围，不会固定绑死进程到某几个核心，操作系统调度器仍会根据系统负载动态调整，性能优于手动固定绑定。

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