python - Python 多处理模块中 apply() 和 apply_async() 的区别

Question

我目前有一段代码可以生成多个进程，如下所示:

pool = Pool(processes=None)
results = [pool.apply(f, args=(arg1, arg2, arg3)) for arg3 in arg_list]

我的想法是，这将跨核心划分工作，使用自 processes=None 以来可用的所有核心。但是，multiprocessing module docs 中的 Pool.apply() 方法的文档阅读:

Equivalent of the apply() built-in function. It blocks until the result is ready, so apply_async() is better suited for performing work in parallel. Additionally, func is only executed in one of the workers of the pool.

第一个问题:我不清楚这一点。 apply 如何在 worker 之间分配工作，它与 apply_async 有何不同？如果任务分布在 worker 中，func 怎么可能只在其中一个 worker 中执行？

我的猜测:我的猜测是，在我当前的实现中，apply 将任务交给具有特定参数集的工作人员，然后等待那个 worker 要完成的工作，然后将下一组参数交给另一个 worker 。通过这种方式，我将工作发送到不同的进程，但没有发生并行性。这似乎是因为 apply 实际上只是:

def apply(self, func, args=(), kwds={}):
    '''
    Equivalent of `func(*args, **kwds)`.
    Pool must be running.
    '''
    return self.apply_async(func, args, kwds).get()

第二个问题:我还想更好地理解为什么，在 docs 的介绍中, 第 16.6.1.5 节。 ('Using a pool of workers')，他们说即使是带有 apply_async 的构造，例如 [pool.apply_async(os.getpid, ()) for i in range(4)]可能使用更多进程，但不确定是否会。 什么决定是否使用多进程？

Answer 1

您已指向 Python2.7 文档，因此我将根据 Python2.7 多处理实现来回答。它在 Python3.X 上可能有所不同，但应该不会有太大差异。

apply和apply_async的区别

当您查看它们的实际实现方式时，这两者之间的区别实际上是 self 描述的。在这里，我将从 multiprocessing/pool.py 中复制/粘贴机器人功能的代码。

def apply(self, func, args=(), kwds={}):
    '''
    Equivalent of `apply()` builtin
    '''
    assert self._state == RUN
    return self.apply_async(func, args, kwds).get()

如您所见，apply 实际上是在调用 apply_async 但就在返回结果之前，调用了 get。这基本上使 apply_async 阻塞，直到返回结果。

def apply_async(self, func, args=(), kwds={}, callback=None):
    '''
    Asynchronous equivalent of `apply()` builtin
    '''
    assert self._state == RUN
    result = ApplyResult(self._cache, callback)
    self._taskqueue.put(([(result._job, None, func, args, kwds)], None))
    return result

apply_async 将任务排入任务队列，并返回提交任务的句柄。使用该 handle，您可以分别调用 get 或 wait 来获取结果或等待任务完成。任务完成后，它返回的内容将作为参数传递给 callback 函数。

示例:

from multiprocessing import Pool
from time import sleep


def callback(a):
    print a


def worker(i, n):
    print 'Entering worker ', i
    sleep(n)
    print 'Exiting worker'
    return 'worker_response'


if __name__ == '__main__':
    pool = Pool(4)
    a = [pool.apply_async(worker, (i, 4), callback=callback) for i in range(8)]
    for i in a:
        i.wait()

结果:

Entering worker  0
Entering worker  1
Entering worker  2
Entering worker  3
Exiting worker
Exiting worker
Exiting worker
Exiting worker
Entering worker  4
Entering worker  5
worker_response
Entering worker  6
worker_response
Entering worker  7
worker_response
worker_response
Exiting worker
Exiting worker
Exiting worker
Exiting worker
worker_response
worker_response
worker_response
worker_response

请注意，使用apply_async时，您必须等待结果或等待任务完成。如果您不对我的示例的最后两行进行注释，您的脚本将在您运行后立即完成。

为什么apply_async可能使用更多的进程

关于 apply 的描述和工作方式，我理解这一点。由于 apply 通过任务将任务发送到 Pool 中的可用进程，apply_async 将任务添加到队列，然后任务队列线程将它们发送到可用进程Pool 中的进程。这就是为什么当您使用 apply_async 时可能会运行多个进程。

我经历了this部分几次，以更好地理解作者试图传达的想法。让我们在这里检查一下:

# evaluate "os.getpid()" asynchronously
res = pool.apply_async(os.getpid, ()) # runs in *only* one process
print res.get(timeout=1)              # prints the PID of that process

# launching multiple evaluations asynchronously *may* use more processes
multiple_results = [pool.apply_async(os.getpid, ()) for i in range(4)]
print [res.get(timeout=1) for res in multiple_results]

如果我们试图通过查看上一个示例来理解上一个示例，当您多次连续调用 apply_async 时，它肯定可能同时运行更多的调用。这可能取决于当时使用了 Pool 中的进程数。这就是为什么他们说可能。