Python并发.futures使用子进程，运行多个python脚本

Question

我想使用并发.futures 同时运行多个 python 脚本。我的代码的串行版本在文件夹中查找特定的 python 文件并执行它。

import re
import os
import glob
import re
from glob import glob
import concurrent.futures as cf

FileList = [];
import time
FileList = [];
start_dir = os.getcwd();
pattern   = "Read.py"

for dir,_,_ in os.walk(start_dir):
    FileList.extend(glob(os.path.join(dir,pattern))) ;

FileList

i=0
for file in FileList:
    dir=os.path.dirname((file))
    dirname1 = os.path.basename(dir) 
    print(dirname1)
    i=i+1
    Str='python '+ file
    print(Str)
    completed_process = subprocess.run(Str)`

对于我的代码的并行版本:

    def Python_callback(future):
    print(future.run_type, future.jid)
    return "One Folder finished executing"

def Python_execute():
    from concurrent.futures import ProcessPoolExecutor as Pool
    args = FileList
    pool = Pool(max_workers=1)
    future = pool.submit(subprocess.call, args, shell=1)
    future.run_type = "run_type"
    future.jid = FileList
    future.add_done_callback(Python_callback)
    print("Python executed")

if __name__ == '__main__':
    import subprocess
    Python_execute()

问题是我不确定如何将 FileList 的每个元素传递给单独的 cpu

感谢您提前提供帮助

Answer 1

最小的变化是使用 submit每个元素一次，而不是整个列表一次:

futures = []
for file in FileList:
    future = pool.submit(subprocess.call, file, shell=1)
    future.blah blah
    futures.append(future)

futures仅当您想对 future 执行某些操作时才需要 list - 等待它们完成、检查它们的返回值等。

同时，您正在使用 max_workers=1 显式创建池。毫不奇怪，这意味着您只会获得 1 个工作子进程，因此它最终会等待一个子进程完成，然后再获取下一个子进程。如果您想同时运行它们，请删除 max_workers并让它默认为每个核心一个(或者传递 max_workers=8 或其他不是 1 的数字，如果您有充分的理由覆盖默认值)。

<小时/>

当我们这样做时，有很多方法可以简化您正在做的事情:

• 你真的需要multiprocessing这里？如果您需要与每个子进程进行通信，那么在单个线程中执行此操作可能会很痛苦 - 但是线程，或者也许 asyncio ，将与此处的流程一样有效。
• 更重要的是，看起来您实际上不需要任何东西，只需启动进程并等待它完成即可，这可以通过简单的同步代码来完成。
• 为什么要构建字符串并使用 shell=1而不是只传递一个列表而不使用 shell？使用 shell 会带来不必要的开销、安全问题和调试烦恼。
• 你真的不需要 jid在每个 future 上 - 它只是所有调用字符串的列表，这没有用。可能更有用的是某种标识符，或者子进程返回代码，或者......可能还有很多其他东西，但它们都是可以通过读取 subprocess.call 的返回值来完成的事情。或一个简单的包装。
• 您确实也不需要回调。如果您只是将所有 future 收集在一个列表中并且 as_completed有了它，您可以更简单地打印显示的结果。
• 如果您执行了上述两项操作，那么除了 pool.submit 之外，您将一无所有。在循环内部 - 这意味着您可以将整个循环替换为 pool.map .
• 您很少需要或想要混合 os.walk和glob 。当您实际上有一个 glob 模式时，请应用 fnmatch超过files列表来自os.walk 。但在这里，您只是在每个目录中查找特定的文件名，所以实际上，您需要过滤的是 file == 'Read.py' .
• 您没有使用i在你的循环中。但如果你确实需要它，最好这样做 for i, file in enumerate(FileList):比做for file in FileList:并手动增加 i .