Python multiprocessing.Queue 与 multiprocessing.manager().Queue()

Question

我有这样一个简单的任务:

def worker(queue):
    while True:
        try:
            _ = queue.get_nowait()
        except Queue.Empty:
            break

if __name__ == '__main__':
    manager = multiprocessing.Manager()
    # queue = multiprocessing.Queue()
    queue = manager.Queue()

    for i in range(5):
        queue.put(i)

    processes = []

    for i in range(2):
        proc = multiprocessing.Process(target=worker, args=(queue,))
        processes.append(proc)
        proc.start()

    for proc in processes:
        proc.join()

似乎 multiprocessing.Queue 可以完成我需要的所有工作，但另一方面，我看到许多 manager().Queue() 的示例，无法理解我真正需要什么。看起来像 Manager().Queue() 使用某种代理对象，但我不明白这些目的，因为 multiprocessing.Queue() 在没有任何代理对象的情况下做同样的工作。

所以，我的问题是:

1) multiprocessing.Queue 和 multiprocessing.manager().Queue() 返回的对象有什么区别？

2) 我需要使用什么？

Answer 1

虽然我对这个主题的理解有限，但从我所做的事情来看，multiprocessing.Queue() 和 multiprocessing.Manager().Queue() 之间存在一个主要区别:

• multiprocessing.Queue() 是一个对象，而 multiprocessing.Manager().Queue() 是一个地址(代理)，指向由 multiprocessing.Manager() 对象管理的共享队列。
• 因此您不能将普通的 multiprocessing.Queue() 对象传递给 Pool 方法，因为它不能被腌制。
• 此外，python doc告诉我们在使用 multiprocessing.Queue() 时要特别注意，因为它可能会产生不良影响

Note When an object is put on a queue, the object is pickled and a background thread later flushes the pickled data to an underlying pipe. This has some consequences which are a little surprising, but should not cause any practical difficulties – if they really bother you then you can instead use a queue created with a manager. After putting an object on an empty queue there may be an infinitesimal delay before the queue’s empty() method returns False and get_nowait() can return without raising Queue.Empty. If multiple processes are enqueuing objects, it is possible for the objects to be received at the other end out-of-order. However, objects enqueued by the same process will always be in the expected order with respect to each other.

Warning As mentioned above, if a child process has put items on a queue (and it has not used JoinableQueue.cancel_join_thread), then that process will not terminate until all buffered items have been flushed to the pipe. This means that if you try joining that process you may get a deadlock unless you are sure that all items which have been put on the queue have been consumed. Similarly, if the child process is non-daemonic then the parent process may hang on exit when it tries to join all its non-daemonic children. Note that a queue created using a manager does not have this issue.

通过将队列设置为全局变量并在初始化时为所有进程设置它，可以将 multiprocessing.Queue() 与 Pool 一起使用:

queue = multiprocessing.Queue()
def initialize_shared(q):
    global queue
    queue=q

pool= Pool(nb_process,initializer=initialize_shared, initargs(queue,))

将创建具有正确共享队列的池进程，但我们可以争辩说 multiprocessing.Queue() 对象不是为此用途而创建的。

另一方面，manager.Queue() 可以在池子进程之间共享，方法是将它作为函数的普通参数传递。

在我看来，使用 multiprocessing.Manager().Queue() 在任何情况下都很好，而且麻烦更少。使用经理可能会有一些缺点，但我不知道。