python - 如何在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”中等待，因为队列中的 elem 太多，并且没有消费者为新 elem 腾出更多空间。

如文档 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"))