如何用Python实现固定输入文件名的外部串行软件多实例并行运行

问题根源 #

• 所有软件实例都读取同一路径的固定名称输入文件，你启动下一个进程前覆盖更新该文件，会导致之前启动、还未完成输入读取的进程拿到错误/不完整的内容，最终仅最后更新输入对应的实例可以正常运行
• 你当前使用multiprocessing.Process属于冗余设计，需要运行的是外部独立程序，直接管理subprocess创建的子进程即可，不需要额外开启Python子进程

解决方案 #

核心思路 #

给每个软件实例创建独立的工作目录，每个目录下存放对应实例的同名输入文件，启动实例时指定专属工作目录，从根源避免输入文件、日志文件互相覆盖的冲突。

代码实现 #

from subprocess import Popen, PIPE
import os
from concurrent.futures import ThreadPoolExecutor, wait

# 保留你原来的全局变量配置
PROGRAM_EXEC = "替换为你的外部可执行程序路径"
SOFTWARE_INPUT_FILE = "固定的输入文件名"

def run_single_instance(instance_setting, instance_id):
    # 1. 创建当前实例专属工作目录
    work_dir = f"./run_task_{instance_id}"
    os.makedirs(work_dir, exist_ok=True)
    
    # 2. 生成当前实例的输入文件，写入专属目录，文件名保持软件要求的固定名
    input_path = os.path.join(work_dir, SOFTWARE_INPUT_FILE)
    # 替换为你原来的prepare_software_input、update_input逻辑，直接写input_path即可
    generate_your_input(input_path, instance_setting)
    
    # 3. 启动外部程序，指定工作目录，程序会自动读取当前目录下的同名输入文件
    p = Popen(
        [PROGRAM_EXEC, SOFTWARE_INPUT_FILE],
        cwd=work_dir, # 核心参数：指定运行工作目录
        stdin=PIPE, stdout=PIPE, stderr=PIPE
    )
    p.communicate()
    # 可自行扩展执行结果判断、日志输出逻辑
    return instance_id, p.returncode

def run_parallel_serial(n_cpu, settings):
    # max_workers设为你要分配的CPU数，即可同时启动对应数量的实例
    with ThreadPoolExecutor(max_workers=n_cpu) as executor:
        futures = []
        for idx in range(n_cpu):
            futures.append(executor.submit(run_single_instance, settings[idx], idx+1))
        # 等待所有实例运行完成再返回
        wait(futures)
    print("所有实例运行完成！")
    return

方案说明 #

• 不需要设置10秒的启动间隔，所有实例可以同时启动，输入、日志都存储在各自的专属目录下，完全不会互相干扰
• 相比原multiprocessing方案更轻量，仅用线程管理外部子进程，没有额外的Python进程开销
• 你需要分配X个CPU并行，只需要把n_cpu参数设为X即可

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