我如何在 Python 中正确使用多进程

Question

我有一个在 Python 中苦苦挣扎的简单示例问题。我正在使用多进程来执行一个示例，其中函数“Thread_Test()”将在 0 到 1 的区间内生成一个统一的随机数数组，数组中的数据点数为“Sample_Size”。一旦我完成了这个示例，我计划生成多个进程副本以尝试加速代码执行，然后我将在 Thread_Test() 中放置一组更复杂的计算。只要我将 Sample_Size 保持在 9,000 以下，此示例就可以正常工作。当我将 Sample_Size 从 10 增加到 8,000 时，执行时间会增加，但在 8,000 时，代码只需要 0.01 秒即可执行。但是，只要我将 Sample_Size 增加到 9,000，代码就会永远执行下去，永远不会完成计算。这是什么原因造成的？

from multiprocessing import Process, Queue
import queue
import random
import timeit
import numpy as np

def Thread_Test(Sample_Size):
    q.put(np.random.uniform(0,1,Sample_Size))
    return

if __name__ == "__main__":
    Sample_Size = 9000
    q = Queue()
    start = timeit.default_timer()
    p = Process(target=Thread_Test,args=(Sample_Size,))
    p.start()
    p.join()

    result = np.array([])
    while True:
        if not q.empty():
         result = np.append(result,q.get())
        else:
           break
    print (result)

    stop = timeit.default_timer()
    print ('{}{:4.2f}{}'.format("Computer Time: ", stop-start, " seconds"))

Answer 1

这个问题的发生是因为如果你把某物放在队列中（你看到的生产者）在子进程中，你必须保证主进程（消费者）同时获取元素。否则，主进程将在“p.join()”中等待，而子进程将在“Queue.put”中等待，因为队列中的元素太多，没有消费者为新元素腾出更多空间。

作为文档here:

Bear in mind that a process that has put items in a queue will wait before terminating until 
all the buffered items are fed by the “feeder” thread to the underlying pipe

简单来说，你需要在“p.join()”之前调用“get part”。

如果您担心主进程在子进程工作之前退出，您可以将代码更改为如下所示：

while True:
    # check subprocess running before break
    running = p.is_alive()
    if not q.empty():
        result = np.append(result,q.get())
    else:
        if not running:
            break

整个部分如下：

def Thread_Test(q, Sample_Size):
    q.put(np.random.uniform(0,1,Sample_Size))


if __name__ == "__main__":
    Sample_Size = 9000
    q = Queue()
    start = timeit.default_timer()
    p = Process(target=Thread_Test,args=(q, Sample_Size,))
    p.daemon = True
    p.start()

    result = np.array([])
    while True:
        running = p.is_alive()
        if not q.empty():
            result = np.append(result,q.get())
        else:
            if not running:
                break
    p.join()
    stop = timeit.default_timer()
    print ('{}{:4.2f}{}'.format("Computer Time: ", stop-start, " seconds"))