multiprocessing.Pool.apply和multiprocessing.Pool.apply_async的用途

Question

参见下面的示例和执行结果：

#!/usr/bin/env python3.4
from multiprocessing import Pool
import time
import os

def initializer():
    print("In initializer pid is {} ppid is {}".format(os.getpid(),os.getppid()))

def f(x):
    print("In f pid is {} ppid is {}".format(os.getpid(),os.getppid()))
    return x*x

if __name__ == '__main__':
    print("In main pid is {} ppid is {}".format(os.getpid(), os.getppid()))
    with Pool(processes=4, initializer=initializer) as pool:  # start 4 worker processes
        result = pool.apply(f, (10,)) # evaluate "f(10)" in a single process
        print(result)

        #result = pool.apply_async(f, (10,)) # evaluate "f(10)" in a single process
        #print(result.get())

给予：

$ ./pooleg.py
In main pid is 22783 ppid is 19542
In initializer pid is 22784 ppid is 22783
In initializer pid is 22785 ppid is 22783
In initializer pid is 22787 ppid is 22783
In f pid is 22784 ppid is 22783
In initializer pid is 22786 ppid is 22783
100

从输出中可以清楚地看到:创建了4个进程，但实际上只有其中一个进程完成了这项工作(称为f)。

问:为什么我要创建一个由>1名工作人员组成的池并调用apply()，而工作f仅由一个进程完成？对于apply_async()来说也是一样的，因为在这种情况下，工作只由一个工作人员完成。

我不明白在哪些用例中这些功能是有用的。

Answer 1

首先，这两种方法都是对参数元组(单函数调用)进行操作，这与可迭代的Pool.map变量相反。因此，当您只观察到只调用这些函数一次时只使用一个进程时，这并不是一个错误。

您可以使用Pool.apply_async，而不是Pool.map版本之一，在该版本中，需要对要分发的单个任务进行更细粒度的控制。

Pool.map版本采用可迭代性，并将它们块到任务中，其中每个任务都具有相同的(映射)目标函数。Pool.apply_async通常不会只被调用一次，它的员工池为>1。由于它是异步的，所以您可以遍历手动预捆绑的任务，并在它们完成之前将它们提交给多个工作进程。您的任务列表可以由不同的目标函数组成，就像您在这个答案here中看到的那样。它还允许为结果和错误注册回调，比如在this示例中。

这些属性使Pool.apply_async非常通用，并且是unusual问题场景的首选工具，您无法使用Pool.map版本之一完成这些工作。

事实上，Pool.apply并不是一见钟情(和第二)就能广泛使用的。您可以使用它来同步控制流，在这样的场景中，首先用apply_async启动多个任务，然后使用apply_async启动另一轮任务之前必须完成一个任务。

使用Pool.apply也意味着为中间任务创建一个额外的进程，而此时您已经有了一个当前空闲的池。

Answer 2

代码中的这一行：

Pool(processes=4, initializer=initializer) as pool:  # start 4 worker processes

不启动4辅助进程。它只是创建了一个池，无法支持同时运行其中的许多。实际上是像apply()这样的方法开始单独运行进程。

区别在于，apply()和apply_async()是前者在结果准备就绪之前阻塞，而后者则立即返回一个“结果”对象。除非您希望一次向Pool提交多个任务(当然，这就是使用multiprocessing模块的全部意义)，否则这不会有太大影响。

下面是对代码的一些修改，展示了如何使用Pool进行一些并发处理

from multiprocessing import Pool
import time
import os

def initializer():
    print("In initializer pid is {} ppid is {}".format(os.getpid(),os.getppid()))

def f(x):
    print("In f pid is {} ppid is {}".format(os.getpid(),os.getppid()))
    return x*x

if __name__ == '__main__':
    print("In main pid is {} ppid is {}".format(os.getpid(), os.getppid()))
    with Pool(processes=4, initializer=initializer) as pool:  # Create 4 worker Pool.
#        result = pool.apply(f, (10,)) # evaluate "f(10)" in a single process
#        print(result)
        # Start multiple tasks.
        tasks = [pool.apply_async(f, (val,)) for val in range(10, 20)]
        pool.close()  # No more tasks.
        pool.join()  # Wait for all tasks to finish.
        results = [result.get() for result in tasks]  # Get the result of each.
        print(results)

map_sync()更适合处理这样的事情(一个值序列)，因为它将自动处理上面代码中显示的一些细节。