Python多进程压测24核CPU遇瓶颈：仅8核满载如何解决？

问题背景 #

在学习Python multiprocessing模块时，我编写了CPU压测代码，目标是让CPU全核心满载。在8核i7-9700F的Windows 11 Home 24H2设备上测试正常，所有核心均能达到100%占用；但在24核i7-12850HX的Windows 10 Enterprise 22H2设备上，仅有8核满载，需要找到将16核或全部24核拉满的方案。

测试代码 #

import multiprocessing
import time

def get_cpu_core_count():
    return multiprocessing.cpu_count()

def stress_core():
    while True:
        pass

if __name__ == '__main__':
    num_cores = int(input("Enter the number of CPU cores to stress: "))
    processes = []

    for _ in range(num_cores):
        p = multiprocessing.Process(target=stress_core)
        p.start()
        processes.append(p)

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        for p in processes:
            p.terminate()
            p.join()

问题根源 #

i7-12850HX采用大小核架构（8个性能核P-Core + 16个能效核E-Core），Windows默认调度策略会优先将Python进程分配到性能核，导致能效核未被充分利用。此外，Python进程的亲和性设置、系统电源计划、Python版本（32位/64位）也可能影响核心识别与调度。

解决办法 #

1. 绑定进程到指定核心（核心方案）

使用psutil库为每个压测进程设置CPU亲和性，强制绑定到不同核心（包括能效核），确保每个核心都被占用。

修改后的代码：

import multiprocessing
import time
import os
import psutil

def stress_core(core_id):
    # 将当前进程绑定到指定核心
    current_process = psutil.Process(os.getpid())
    current_process.cpu_affinity([core_id])
    while True:
        pass

if __name__ == '__main__':
    total_cores = multiprocessing.cpu_count()
    num_cores = int(input(f"Enter the number of CPU cores to stress (max {total_cores}): "))
    processes = []

    for core_idx in range(num_cores):
        p = multiprocessing.Process(target=stress_core, args=(core_idx,))
        p.start()
        processes.append(p)

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        for p in processes:
            p.terminate()
            p.join()

使用前需安装psutil：

pip install psutil

运行时输入24，即可让所有核心满载。

2. 检查系统设置

• 电源计划：切换到「高性能」模式，避免系统因省电限制能效核运行。
• 核心识别：确认multiprocessing.cpu_count()返回值为24，若返回8，需检查：
  • 是否使用64位Python（32位Python最多识别8个核心）；
  • 系统BIOS/设备管理器中是否禁用了部分核心。

3. 调整multiprocessing启动方式

Windows下multiprocessing默认使用spawn启动方式，可尝试改为forkserver（Python 3.4+支持），优化进程创建与调度：

if __name__ == '__main__':
    multiprocessing.set_start_method('forkserver')
    total_cores = multiprocessing.cpu_count()
    # 后续代码不变

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