如何实现Python脚本b.py的多进程并行调用以缩短执行耗时？

嘿，这个场景太常见了！串行跑4个b.py要4分钟，并行起来理论上能压缩到1分钟左右，效率提升明显。下面给你几种Python原生的解决方案，都是不用额外装包就能用的：

方案1：用concurrent.futures.ProcessPoolExecutor（最推荐，代码简洁易维护） #

这个是Python3.2+自带的高级API，帮你自动管理进程池，不用手动处理进程的创建和销毁。修改后的a.py如下：

import subprocess
from concurrent.futures import ProcessPoolExecutor

if __name__ == '__main__':
    inputs = ['file1', 'file2', 'file3', 'file4']
    script_path = 'b.py'

    def run_bpy(infile):
        # 用subprocess.run替代os.system，更安全可控，能捕获执行状态
        try:
            # 如果需要把每个b.py的日志单独存，就打开下面的注释，重定向输出到文件
            # with open(f'b_log_{infile}.txt', 'w') as log_file:
            #     subprocess.run([script_path, '-i', infile], stdout=log_file, stderr=log_file, check=True)
            subprocess.run([script_path, '-i', infile], check=True)
            print(f"✅ 处理完成：{infile}")
        except subprocess.CalledProcessError as e:
            print(f"❌ 处理失败：{infile}，错误码：{e.returncode}")

    # max_workers可以根据你的CPU核心数调整，比如用multiprocessing.cpu_count()获取核心数
    with ProcessPoolExecutor(max_workers=4) as executor:
        executor.map(run_bpy, inputs)

为什么推荐这个？ #

• 代码简洁，不用手动管理进程生命周期
• 自带异常处理机制，能轻松捕获单个b.py执行失败的情况
• 可以灵活控制并发数，避免系统资源过载

方案2：用multiprocessing模块（手动控制进程，更灵活） #

如果你需要更精细地控制每个进程的状态，比如单独监控某个进程是否完成，可以用这个原生的多进程模块：

import subprocess
import multiprocessing

if __name__ == '__main__':
    inputs = ['file1', 'file2', 'file3', 'file4']
    script_path = 'b.py'

    def run_bpy(infile):
        try:
            subprocess.run([script_path, '-i', infile], check=True)
            print(f"✅ 处理完成：{infile}")
        except subprocess.CalledProcessError as e:
            print(f"❌ 处理失败：{infile}，错误码：{e.returncode}")

    # 创建并启动所有进程
    processes = []
    for infile in inputs:
        proc = multiprocessing.Process(target=run_bpy, args=(infile,))
        processes.append(proc)
        proc.start()

    # 等待所有进程执行完毕
    for proc in processes:
        proc.join()

方案3：用subprocess.Popen（最轻量化的方式） #

如果不想引入多进程模块，直接用subprocess.Popen也能实现并行，它会直接启动子进程而不等待执行完成：

import subprocess

if __name__ == '__main__':
    inputs = ['file1', 'file2', 'file3', 'file4']
    script_path = 'b.py'

    procs = []
    for infile in inputs:
        # 启动子进程，立即返回进程对象，不阻塞
        proc = subprocess.Popen([script_path, '-i', infile])
        procs.append(proc)
        print(f"🚀 启动进程处理：{infile}")

    # 逐个等待所有进程结束
    for proc in procs:
        return_code = proc.wait()
        if return_code == 0:
            print(f"✅ 进程完成，返回码：{return_code}")
        else:
            print(f"❌ 进程失败，返回码：{return_code}")

几个关键注意事项： #

1. 控制并发数：如果你的inputs列表很长（比如几十上百个），不要把并发数设得太大，建议和你的CPU核心数一致（用multiprocessing.cpu_count()获取），避免系统资源被占满导致卡顿。
2. 日志隔离：并行执行时多个b.py的控制台输出会混在一起，非常难排查问题，强烈建议把每个b.py的日志重定向到单独的文件（参考方案1里的注释代码）。
3. 资源冲突：如果b.py需要读写同一个文件、访问共享数据库等，一定要加锁或者确保资源是进程安全的，否则会出现数据错乱的问题。
4. 错误重试：如果某些文件处理失败需要重试，可以在run_bpy函数里加循环重试逻辑，或者收集失败的任务后续再处理。

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