python multithreading等到所有线程都完成了

Question

这可能是在类似的背景下提出的,但是在搜索约20分钟后我无法找到答案,所以我会问.

我编写了一个Python脚本(比方说:scriptA.py)和一个脚本(比如说scriptB.py)

在scriptB中我想用不同的参数多次调用scriptA,每次运行大约需要一个小时,(它是一个巨大的脚本,做了很多东西......不用担心它)我希望能够运行scriptA同时具有所有不同的参数,但我需要等到所有这些都完成后再继续; 我的代码:

import subprocess

#setup
do_setup()

#run scriptA
subprocess.call(scriptA + argumentsA)
subprocess.call(scriptA + argumentsB)
subprocess.call(scriptA + argumentsC)

#finish
do_finish()

我想同时运行所有subprocess.call(),然后等到它们全部完成,我该怎么做？

我试着像这里的例子一样使用线程:

from threading import Thread
import subprocess

def call_script(args)
    subprocess.call(args)

#run scriptA   
t1 = Thread(target=call_script, args=(scriptA + argumentsA))
t2 = Thread(target=call_script, args=(scriptA + argumentsB))
t3 = Thread(target=call_script, args=(scriptA + argumentsC))
t1.start()
t2.start()
t3.start()

但我不认为这是对的.

在我去之前我怎么知道他们都跑完了do_finish()？

Answer 1

将线程放在列表中,然后使用Join方法

 threads = []

 t = Thread(...)
 threads.append(t)

 ...repeat as often as necessary...

 # Start all threads
 for x in threads:
     x.start()

 # Wait for all of them to finish
 for x in threads:
     x.join()

Answer 2

您需要在脚本的末尾使用object的join方法Thread.

t1 = Thread(target=call_script, args=(scriptA + argumentsA))
t2 = Thread(target=call_script, args=(scriptA + argumentsB))
t3 = Thread(target=call_script, args=(scriptA + argumentsC))

t1.start()
t2.start()
t3.start()

t1.join()
t2.join()
t3.join()

因此主线程将等待t1,t2并t3完成执行.

Answer 3

我更喜欢使用基于输入列表的列表推导:

inputs = [scriptA + argumentsA, scriptA + argumentsB, ...]
threads = [Thread(target=call_script, args=(i)) for i in inputs]
[t.start() for t in threads]
[t.join() for t in threads]

Answer 4

在Python3中,由于Python 3.2有一种新的方法可以达到相同的结果,我个人更喜欢传统的线程创建/启动/加入,包concurrent.futures:https://docs.python.org/3/library/concurrent.futures html的

使用ThreadPoolExecutor代码将是:

from concurrent.futures.thread import ThreadPoolExecutor
import time

def call_script(ordinal, arg):
    print('Thread', ordinal, 'argument:', arg)
    time.sleep(2)
    print('Thread', ordinal, 'Finished')

args = ['argumentsA', 'argumentsB', 'argumentsC']

with ThreadPoolExecutor(max_workers=2) as executor:
    ordinal = 1
    for arg in args:
        executor.submit(call_script, ordinal, arg)
        ordinal += 1
print('All tasks has been finished')

前面代码的输出类似于:

Thread 1 argument: argumentsA
Thread 2 argument: argumentsB
Thread 1 Finished
Thread 2 Finished
Thread 3 argument: argumentsC
Thread 3 Finished
All tasks has been finished

其中一个优点是您可以控制吞吐量设置最大并发工作者.

Answer 5

你可以有类类似下面从中你可以添加你想要并行激情执行,并开始执行,并等待所有作业完成的功能或console_scripts"N"数字..

from multiprocessing import Process

class ProcessParallel(object):
    """
    To Process the  functions parallely

    """    
    def __init__(self, *jobs):
        """
        """
        self.jobs = jobs
        self.processes = []

    def fork_processes(self):
        """
        Creates the process objects for given function deligates
        """
        for job in self.jobs:
            proc  = Process(target=job)
            self.processes.append(proc)

    def start_all(self):
        """
        Starts the functions process all together.
        """
        for proc in self.processes:
            proc.start()

    def join_all(self):
        """
        Waits untill all the functions executed.
        """
        for proc in self.processes:
            proc.join()


def two_sum(a=2, b=2):
    return a + b

def multiply(a=2, b=2):
    return a * b


#How to run:
if __name__ == '__main__':
    #note: two_sum, multiply can be replace with any python console scripts which
    #you wanted to run parallel..
    procs =  ProcessParallel(two_sum, multiply)
    #Add all the process in list
    procs.fork_processes()
    #starts  process execution 
    procs.start_all()
    #wait until all the process got executed
    procs.join_all()