如何并行运行带有参数的函数？

Question

import timefrom multiprocessing import Processdef worker():    time.sleep(2)    print("Working")def runInParallel(*fns):    proc = []    for fn in fns:        p = Process(target=fn)        p.start()        proc.append(p)    for p in proc:        p.join()if __name__ == '__main__':    start = time.time()    runInParallel(worker, worker, worker, worker)    print("Total time taken: ", time.time()-start)但是，如果我添加参数，worker()它就不再并行运行。import timefrom multiprocessing import Processdef worker(ii):    time.sleep(ii)    print("Working")def runInParallel(*fns):    proc = []    for fn in fns:        p = Process(target=fn)        p.start()        proc.append(p)    for p in proc:        p.join()if __name__ == '__main__':    start = time.time()    runInParallel(worker(2), worker(2), worker(2), worker(2))    print("Total time taken: ", time.time()-start)原因可能是什么？

Answer 1

您应该修改runInParallel以进行可迭代的解包。

import time

from multiprocessing import Process

def worker(ii):

    time.sleep(ii)

    print("Working")

def runInParallel(*fns):

    proc = []

    for fn in fns:

        func, *args = fn

        p = Process(target=func, args=args)

        p.start()

        proc.append(p)

    for p in proc:

        p.join()

if __name__ == '__main__':

    start = time.time()

    runInParallel((worker, 2), (worker, 3), (worker, 5), (worker, 2))

    print("Total time taken: ", time.time()-start)

Answer 2

worker这是因为和之间的差异worker()。前者是函数，后者是函数调用。线路上发生的情况runInParallel(worker(2), worker(2), worker(2), worker(2))是，在开始执行之前，所有四个调用runInParallel都已运行。如果你print(fns)在开头添加一个，runInParallel你会看到一些不同。

快速解决：

def worker_caller():

    worker(2)

和：

runInParallel(worker_caller, worker_caller, worker_caller, worker_caller)

这不是很方便，但主要是为了显示问题所在。问题不在函数中worker。问题在于您混淆了传递函数和传递函数调用。如果您将第一个版本更改为：

runInParallel(worker(), worker(), worker(), worker())

那么你会遇到完全相同的问题。

但你可以这样做：

runInParallel(lambda:worker(2), lambda: worker(2), lambda: worker(2), lambda: worker(2))

Lambda 非常有用。这是另一个版本：

a = lambda:worker(2)

b = lambda:worker(4)

c = lambda:worker(3)

d = lambda:worker(1)

runInParallel(a, b, c, d)

Answer 3

要传递参数，您需要将它们传递给Process构造函数：

        p = Process(target=fn, args=(arg1,))

Answer 4

Process 构造函数接受 args 和 kwargs 参数，然后在执行时将这些参数传递给进程。文档对此非常清楚。

所以你的代码应该修改如下：

def worker(ii):

    time.sleep(ii)

    print("Working")

def runInParallel(*fns):

    proc = []

    for fn in fns:

        p = Process(target=fn, args=(2,))

        p.start()

        proc.append(p)

    for p in proc:

        p.join()

if __name__ == '__main__':

    start = time.time()

    runInParallel(worker, worker, worker, worker)

    print("Total time taken: ", time.time()-start)

当然，每个进程的参数可能不同，您需要安排将正确的参数传递给 args（或关键字参数的 kwargs）中的每个进程。这可以通过传递元组来实现，例如runInParallel((worker,2), (worker,3), (worker,5), (worker,1)，然后处理内部的元组runInParallel。