Python多进程处理轻量函数效率低，如何模拟多实例提升CPU利用率？

算力阿峰

2026-5-28

轻量函数多进程效率低下的优化方案

问题背景

你现在用Python的multiprocessing.Pool跑数百万次轻量函数，结果每个CPU核心只用到30%左右，只有锁操作时才占满。性能分析显示锁和多进程机制的开销是大头，并行跑要15分钟，串行反而只要4.5分钟。而且你的函数都是写入独立文件，完全没有依赖，想知道能不能通过启动多个独立并行脚本的方式充分利用CPU。

当前代码

初始版本（带tqdm锁）：

pool = Pool(cpu_count(), initializer=tqdm.tqdm.set_lock, initargs=(Lock(),)) 
for _ in tqdm.tqdm(pool.imap_unordered(parallel_process, pubfiles, chunksize=70), total=nfiles, desc='Parsing files'): 
    pass

修改后版本（去掉tqdm锁）：

pool = Pool(cpu_count()) 
for i in pool.imap_unordered(parallel_process, files, chunksize=70): 
    print(i)

核心原因

你的函数太轻量了！multiprocessing.Pool的任务调度、进程间通信（哪怕是传递简单的文件名）的开销，都比单个函数的执行时间大得多。再加上锁操作的额外消耗，直接导致并行的“收益”抵不上“开销”，反而比串行慢。

优化方案

1. 优先调整：合并小任务，摊平调度开销

这是最简单有效的方法——把多个小任务打包成一个批次，让每个进程一次处理一批文件，而不是单个文件。这样能大幅减少进程间的任务分配次数，把调度开销摊到更多任务上。

比如把chunksize调大N倍（比如从70改成7000，根据总文件数调整），或者手动打包任务：

from multiprocessing import Pool, cpu_count

def batch_process(file_batch):
    # 一个批次里处理多个文件
    for file in file_batch:
        parallel_process(file)

if __name__ == "__main__":
    # 把文件列表分成和CPU核心数相等的批次
    total_files = len(pubfiles)
    batch_size = total_files // cpu_count() + 1
    file_batches = [pubfiles[i:i+batch_size] for i in range(0, total_files, batch_size)]

    with Pool(cpu_count()) as pool:
        # 给每个进程分配一个大批次
        for _ in pool.imap_unordered(batch_process, file_batches):
            pass

这样每个进程一次处理一大组文件，调度开销几乎可以忽略，CPU利用率肯定能上去。

2. 备选：用多线程代替多进程

如果你的函数以IO操作为主（比如写入文件），多线程的开销比多进程小很多——因为线程切换的成本远低于进程，而且文件IO时GIL会自动释放，不会影响并行效率。你可以试试用concurrent.futures.ThreadPoolExecutor代替multiprocessing.Pool：

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=cpu_count()) as executor:
    for _ in executor.map(parallel_process, pubfiles):
        pass

不过如果函数有CPU密集的逻辑，这个方法可能效果不好，但你的场景是函数极轻，值得一试。

3. 你提到的思路：启动多个独立脚本

这个方案完全可行，而且能彻底避免multiprocessing.Pool的调度和锁开销！因为每个独立脚本是一个单独的Python进程，完全独立，没有进程间通信的成本。

具体步骤：

拆分任务：把总文件列表分成N份（N等于CPU核心数），可以把每个子集保存成临时文件，或者直接通过命令行参数传递。
写子脚本：比如写一个process_subset.py，接受子集的索引或者文件路径列表，负责处理自己的那部分文件。
启动多进程：用subprocess或者直接在命令行启动N个python process_subset.py --subset 0这样的进程，每个处理自己的子集。

示例子脚本process_subset.py：

import argparse

def parallel_process(file):
    # 你的原有处理逻辑
    ...

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--files", nargs="+", help="List of files to process")
    args = parser.parse_args()
    
    for file in args.files:
        parallel_process(file)

然后主脚本启动子进程：

import subprocess
from multiprocessing import cpu_count

# 拆分文件列表
total_files = len(pubfiles)
batch_size = total_files // cpu_count() + 1
file_batches = [pubfiles[i:i+batch_size] for i in range(0, total_files, batch_size)]

# 启动多个子进程
processes = []
for batch in file_batches:
    cmd = ["python", "process_subset.py", "--files"] + batch
    proc = subprocess.Popen(cmd)
    processes.append(proc)

# 等待所有进程完成
for proc in processes:
    proc.wait()

这个方法的优点是完全隔离进程，没有Pool的额外开销，每个进程都能跑满CPU；缺点是需要手动处理任务拆分和子进程管理，但对于你的独立任务场景来说，成本很低。

4. 额外优化：去掉不必要的锁和监控

你已经去掉了tqdm的锁，但还要检查parallel_process函数里有没有其他不必要的锁——比如全局变量锁、不必要的文件锁（你说写入独立文件，应该不需要）。另外，如果不需要实时进度，完全去掉tqdm的进度条，能进一步减少开销。

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

Python多进程处理轻量函数效率低，如何模拟多实例提升CPU利用率？

问题背景 #

当前代码 #

核心原因 #

优化方案 #