如何将使用GNU Parallel并行处理文件的Shell脚本重写为Python代码

嗨，这个问题我熟！把GNU Parallel那套并行逻辑转到Python里其实不难，结合你已经写好的run函数，我给你拆解几个靠谱的实现方案，还会解决类似--linebuffer的实时输出问题：

GNU Parallel本质是启动多个独立进程并行处理任务，Python里对应的就是进程池（避开GIL限制，完美适配调用外部程序或CPU密集型任务的场景），同时我们要解决实时输出的问题，对应原命令里的--linebuffer。

原来的os.command已经被官方废弃，而且没法精准控制输出缓冲，建议换成subprocess模块，这样能完美实现类似--linebuffer的实时输出效果：

import subprocess
import sys

def run(file_path):
    # 这里放你的Python预处理代码...
    
    # 替换os.command，用subprocess实现实时行缓冲输出
    with subprocess.Popen(
        ["你的外部程序路径", file_path],  # 替换成实际的外部命令及参数
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,  # 把stderr合并到stdout一起输出
        text=True,
        bufsize=1  # 开启行缓冲，对应GNU Parallel的--linebuffer
    ) as proc:
        # 逐行读取并打印输出，实现实时显示
        for line in proc.stdout:
            print(f"[{file_path}] {line}", end='')
        # 等待程序执行完成，获取返回码
        proc.wait()
    return proc.returncode

这是Python3.2+自带的高层API，用法简洁直观，和GNU Parallel的逻辑最贴近：

from concurrent.futures import ProcessPoolExecutor

def main():
    # 替换成你的实际文件列表
    files = ["file1.txt", "file2.jpg", "file3.dat", ...]
    max_workers = 5  # 对应原命令的--jobs 5

    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        # 批量提交所有文件处理任务
        futures = [executor.submit(run, file) for file in files]
        
        # 遍历任务，等待完成并处理结果/异常
        for future in futures:
            try:
                return_code = future.result()
                # 可以根据返回码做后续处理，比如判断是否执行成功
                if return_code != 0:
                    print(f"任务执行失败，返回码: {return_code}", file=sys.stderr)
            except Exception as e:
                print(f"任务执行出错: {str(e)}", file=sys.stderr)

if __name__ == "__main__":
    main()

如果需要更底层的控制，可以用Python内置的multiprocessing模块的Pool，比如用imap_unordered实时获取已完成的任务结果：

from multiprocessing import Pool
import sys

def main():
    files = ["file1.txt", "file2.jpg", "file3.dat", ...]
    
    with Pool(processes=5) as pool:
        # imap_unordered会按任务完成顺序返回结果，而不是提交顺序
        for result in pool.imap_unordered(run, files):
            if result != 0:
                print(f"有任务执行失败，返回码: {result}", file=sys.stderr)

if __name__ == "__main__":
    main()

• 必须在if __name__ == "__main__":代码块里启动进程池，这是跨平台的要求，能避免进程启动时的递归创建问题。
• 实时输出的核心是subprocess.Popen配合bufsize=1（行缓冲），逐行读取并打印输出，这样每个任务的日志会实时显示，和GNU Parallel的--linebuffer效果完全一致。
• 如果你的run函数需要访问共享资源（比如同一个配置文件、数据库连接），要记得用进程间同步机制（比如multiprocessing.Lock），但如果是每个文件独立处理，就不需要额外处理。

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