Python Multiprocessing调用外部可执行文件避免输入文件覆盖问题问询

多进程下输入文件被覆盖的核心原因是多个进程同时读写同一个文件名，解决的核心思路是让每个进程拥有独立的输入/输出文件，或者完全绕过文件传递数据。下面是几种可行的方案：

1. 为每个任务生成唯一文件名 #

最简单的方式是给每个进程处理的任务分配唯一标识（比如任务索引、进程ID、UUID），以此作为输入/输出文件名的一部分，确保不会重复。

示例代码：

import multiprocessing
import numpy as np
import subprocess
import os

def process_row(row_idx, row_data):
    # 生成唯一文件名
    input_file = f"input_{row_idx}.txt"
    output_file = f"output_{row_idx}.txt"
    
    # 将行数据写入输入文件
    np.savetxt(input_file, [row_data])
    
    # 运行外部可执行文件
    subprocess.run(["./your_exe", input_file, output_file], check=True)
    
    # 读取输出结果
    result = np.loadtxt(output_file)
    
    # 清理临时文件（可选）
    os.remove(input_file)
    os.remove(output_file)
    
    return row_idx, result

if __name__ == "__main__":
    # 模拟输入数据
    input_array = np.random.rand(10, 5)
    
    # 异步并行处理
    with multiprocessing.Pool() as pool:
        results = []
        for idx, row in enumerate(input_array):
            # 提交异步任务
            res = pool.apply_async(process_row, args=(idx, row))
            results.append(res)
        
        # 汇总结果
        final_results = {}
        for res in results:
            idx, data = res.get()
            final_results[idx] = data
        
        # 按原顺序整理结果
        sorted_results = [final_results[i] for i in range(len(input_array))]

2. 使用临时文件模块自动管理独立文件 #

Python的tempfile模块可以帮你创建自动隔离的临时文件/目录，无需手动生成唯一名称，还能在使用后自动清理（或手动控制）。

示例代码（临时文件）：

import multiprocessing
import numpy as np
import subprocess
import tempfile
import os

def process_row(row_data):
    # 创建临时输入文件，delete=False确保外部程序能读取
    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f_in:
        np.savetxt(f_in, [row_data])
        input_file = f_in.name
    
    # 创建临时输出文件
    with tempfile.NamedTemporaryFile(mode='r', delete=False, suffix='.txt') as f_out:
        output_file = f_out.name
    
    try:
        # 运行外部程序
        subprocess.run(["./your_exe", input_file, output_file], check=True)
        # 读取结果
        result = np.loadtxt(output_file)
    finally:
        # 手动清理临时文件
        os.remove(input_file)
        os.remove(output_file)
    
    return result

if __name__ == "__main__":
    input_array = np.random.rand(10, 5)
    
    with multiprocessing.Pool() as pool:
        # 异步映射处理每一行
        sorted_results = pool.map_async(process_row, input_array).get()

如果外部程序需要在特定目录下读写文件，可使用临时目录自动清理：

def process_row(row_data):
    with tempfile.TemporaryDirectory() as temp_dir:
        input_file = os.path.join(temp_dir, "input.txt")
        output_file = os.path.join(temp_dir, "output.txt")
        
        np.savetxt(input_file, [row_data])
        subprocess.run(["./your_exe", input_file, output_file], check=True)
        result = np.loadtxt(output_file)
    
    return result

TemporaryDirectory会在上下文结束后自动删除整个目录，无需手动清理。

3. 绕过文件，通过管道传递输入（推荐如果外部程序支持） #

如果你的外部可执行文件支持从标准输入（stdin）读取数据，并将结果输出到标准输出（stdout），那完全不需要生成文件，直接通过管道传递数据，从根源避免文件冲突。

示例代码：

import multiprocessing
import numpy as np
import subprocess

def process_row(row_data):
    # 将行数据转为符合外部程序要求的字符串格式
    input_str = '\t'.join(map(str, row_data)) + '\n'
    
    # 运行外部程序，通过stdin传递输入，stdout捕获输出
    proc = subprocess.run(
        ["./your_exe"],
        input=input_str.encode('utf-8'),
        capture_output=True,
        text=True,
        check=True
    )
    
    # 解析输出结果（根据外部程序的输出格式调整）
    result = np.fromstring(proc.stdout.strip(), sep='\t')
    return result

if __name__ == "__main__":
    input_array = np.random.rand(10, 5)
    
    with multiprocessing.Pool() as pool:
        sorted_results = pool.map_async(process_row, input_array).get()

这种方式效率最高，没有文件IO的开销，也完全不会有文件覆盖的问题，但前提是外部程序支持标准输入输出。

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