Python正在等待，直到在concurrent.futures中完成许多操作

Question

这就是我尝试使用python和concurrent.futures库的原因

假设您的字典输入为：

input = {'1': {'Joe','Sal', 'Murry','Q}}, {'2':{'Dana','White'}},{'3':{'ice','cream','water','cola','Butter'}}

将在执行器中运行的函数将只接受一个参数。比如：

def printMyInput(mystring):
    print(mystring)

因此，规范是您可以并行运行集合

with concurrent.futures.ProcessPoolExecutor() as executor:
   results = [executor.sumbit(printMyInput, x) for x in input['1'])

这仅对input['1']执行。要运行input['2']，必须首先完成input['1']的进程。对于input['3']，需要完成input['2']的过程，等等。我想不出一个办法来做那件事。我真的需要一些帮助

Answer 1

执行器将阻塞，直到所有进程完成。您可以循环executor调用以按顺序运行每个阶段

请尝试以下代码：

import concurrent.futures

from multiprocessing import Process, freeze_support

input = {'1': ['Joe','Sal', 'Murry','Q'], 
         '2':['Dana','White'],
         '3':['ice','cream','water','cola','Butter']}

def printMyInput(mystring):
    print(mystring)
    
if __name__ == '__main__':
    freeze_support()  # needed for Windows
    for k in input:
       with concurrent.futures.ProcessPoolExecutor() as executor:
          print('>> Process ', k)
          results = [executor.submit(printMyInput, x) for x in input[k]]

输出

>> Process  1
Joe
Sal
Murry
Q
>> Process  2
Dana
White
>> Process  3
ice
cream
water
cola
Butter

更新

为了测试循环结构，我在回调中添加了一个随机休眠来改变处理时间。要使代码正常工作，for必须位于with之上。需要为每个阶段创建一个新的ProcessPoolExecutor

以下是测试代码：

import concurrent.futures
import random, time

from multiprocessing import Process, freeze_support

input = {'1':['Joe','Sal', 'Murry','Q'], 
         '2':['Dana','White'],
         '3':['ice','cream','water','cola','Butter']}

def printMyInput(mystring):
    time.sleep(random.randint(1,4))  # random sleep time
    print(mystring)
    
if __name__ == '__main__':
    freeze_support()  # needed for Windows
    for k in input:
       with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
          print('>> Process ', k)
          results = [executor.submit(printMyInput, x) for x in input[k]]

更新#2

我做了更多的测试。只要在收集结果时阻塞，就可以切换循环结构

import concurrent.futures
import random, time

from multiprocessing import Process, freeze_support

input = {'1':['Joe','Sal', 'Murry','Q'], 
         '2':['Dana','White'],
         '3':['ice','cream','water','cola','Butter']}

def printMyInput(mystring):
    time.sleep(random.randint(1,4))  # random sleep time
    print(mystring)
    
if __name__ == '__main__':
    freeze_support()  # needed for Windows
    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
       for k in input:
          print('>> Process ', k)
          procs = [executor.submit(printMyInput, x) for x in input[k]]
          results = [p.result() for p in procs]  # block until results are gathered
          print('>> Done Process ', k)