函数仅在从 apply 调用时才被评估，而不是在从 apply_async 调用时被评估

Question

我已阅读 process-or-pool-for-what-i-am-doing 和 when-to-use-apply-apply-async-or-map 并希望了解 apply 和 apply_async 之间的区别。但是我有这段代码，它仅在使用 apply 时返回所需的输出，并且在使用 apply_async 时非常短：

#!/bin/env python
import multiprocessing
import time
import os

semaphore = multiprocessing.Semaphore(1)
# semaphore = multiprocessing.Manager().Semaphore(1)

def producer(num, len, output):
    time.sleep(1)
    element = "PROCESS: %d PID: %d, PPID: %d, QSIZE: %d" % (num, os.getpid(), os.getppid(), output.qsize())
    semaphore.acquire()
    print "PID: %s WRITE -> %s" % (os.getpid(), element)
    if (num == len - 1):
        print "PID: %d WRITE -> Everything was written inside queue, no more apply_async calling, just reading!" % os.getpid()
    output.put(element)
    semaphore.release()
    time.sleep(1)

def consumer(output):
    while True:
      try:
        print "PID: %d READ  <- %s" % (os.getpid(), output.get())
        break
      except:
        print "PID: %d READ  <- NOTHING IN BUFFER" % os.getpid()
        # pass
      time.sleep(1)

if __name__ == '__main__':
    """
    MULTIPLE PRODUCERS AND MULTIPLE CONSUMERS
    """
    output    = multiprocessing.Manager().Queue()
    pool      = multiprocessing.Pool(4)
    lst       = range(40)

    print "Calling apply*!"
    for i in lst:
        pool.apply_async(producer, (i, len(lst), output))
    print "Do not wait until apply* finishes!"

    for i in lst:
        # RETURNS OUTPUT
        # pool.apply(consumer, (output,))

        # DOES NOT RETURN OUTPUT
        pool.apply_async(consumer, (output,))

使用pool.apply时的输出：

Calling apply*!
Do not wait until apply* finishes!
PID: 18348 WRITE -> PROCESS: 1 PID: 18348, PPID: 18341, QSIZE: 0
PID: 18346 WRITE -> PROCESS: 0 PID: 18346, PPID: 18341, QSIZE: 1
PID: 18349 WRITE -> PROCESS: 2 PID: 18349, PPID: 18341, QSIZE: 2
PID: 18347 WRITE -> PROCESS: 3 PID: 18347, PPID: 18341, QSIZE: 3
PID: 18346 WRITE -> PROCESS: 4 PID: 18346, PPID: 18341, QSIZE: 4
PID: 18348 WRITE -> PROCESS: 5 PID: 18348, PPID: 18341, QSIZE: 5
PID: 18349 WRITE -> PROCESS: 6 PID: 18349, PPID: 18341, QSIZE: 6
PID: 18347 WRITE -> PROCESS: 7 PID: 18347, PPID: 18341, QSIZE: 7
...

使用pool.apply_async时的输出：

Calling apply*!
Do not wait until apply* finishes!

似乎producer 仅在从 apply 调用时评估，但在从 apply_async 调用时不评估。为什么？

Answer 1

您的代码在任何一种情况下都会被评估，但它是在另一个进程中完成的。区别在于apply 是阻塞的，而apply_async 不是。在您的代码中，您将工作发送到另一个进程，然后再将其收集回主进程。

注意apply 返回一个值，而apply_async 返回一个结果对象。您必须在结果对象上调用 get 才能获得结果。这是一个提炼的例子：

>>> import multiprocessing
>>> import math
>>> 
>>> p = multiprocessing.Pool() 
>>> p.apply(math.sin, (.5,))
0.479425538604203
>>> result = p.apply_async(math.sin, (.5,))
>>> result 
<multiprocessing.pool.ApplyResult object at 0x103edc350>
>>> result.get()
0.479425538604203
>>> 

如果您对apply 或apply_async 执行for 循环，您可能需要考虑使用map 或map_async。

>>> p.map(math.sin, range(5))
[0.0, 0.8414709848078965, 0.9092974268256817, 0.1411200080598672, -0.7568024953079282]
>>> result = p.map_async(math.sin, range(5))
>>> result.get()
[0.0, 0.8414709848078965, 0.9092974268256817, 0.1411200080598672, -0.7568024953079282]