python - python中的多线程系统调用

Question

我有一个类似这样的python脚本:

def test_run():
     global files_dir
     for f1 in os.listdir(files_dir):
          for f2 os.listdir(files_dir):
               os.system("run program x on f1 and f2")

调用每个 os.system 的最佳方式是什么？调用不同的处理器？使用子进程还是多处理池？

注意:程序的每次运行都会生成一个输出文件。

Answer 1

@unutbu 的回答很好，但有一种破坏性较小的方法:使用 Pool传递任务。这样你就不必处理自己的队列了。例如，

import os
NUM_CPUS = None  # defaults to all available

def worker(f1, f2):
    os.system("run program x on f1 and f2")

def test_run(pool):
     filelist = os.listdir(files_dir)
     for f1 in filelist:
          for f2 in filelist:
               pool.apply_async(worker, args=(f1, f2))

if __name__ == "__main__":
     import multiprocessing as mp
     pool = mp.Pool(NUM_CPUS)
     test_run(pool)
     pool.close()
     pool.join()

那“看起来更像”您开始使用的代码。并不是说这一定是件好事;-)

在 Python 3 的最新版本中，Pool对象也可以用作上下文管理器，因此尾端可以简化为:

if __name__ == "__main__":
     import multiprocessing as mp
     with mp.Pool(NUM_CPUS) as pool:
         test_run(pool)

编辑:使用 concurrent.futures 代替

对于像这样的非常简单的任务，Python 3 的 concurrent.futures可以更容易使用。替换上面的代码，来自 test_run()下来，像这样:

def test_run():
     import concurrent.futures as cf
     filelist = os.listdir(files_dir)
     with cf.ProcessPoolExecutor(NUM_CPUS) as pp:
         for f1 in filelist:
             for f2 in filelist:
                 pp.submit(worker, f1, f2)

if __name__ == "__main__":
     test_run()

如果您不希望工作进程中的异常无声地消失，则需要更漂亮。这是所有并行性噱头的潜在问题。问题是通常没有好的方法在主程序中引发异常，因为它们发生在可能与主程序当时正在做的事情无关的上下文(工作进程)中。在主程序中获取(重新)引发异常的一种方法是明确要求结果；例如，将上面的更改为:

def test_run():
     import concurrent.futures as cf
     filelist = os.listdir(files_dir)
     futures = []
     with cf.ProcessPoolExecutor(NUM_CPUS) as pp:
         for f1 in filelist:
             for f2 in filelist:
                 futures.append(pp.submit(worker, f1, f2))
     for future in cf.as_completed(futures):
         future.result()

那么如果工作进程发生异常，future.result()当该异常应用于 Future 时，将在主程序中重新引发该异常。表示失败的进程间调用的对象。

在这一点上可能比你想知道的要多;-)