时间戳

首页 > 解决方案 > 如何使用 multiprocessing.Queue.get 方法？

python - 如何使用 multiprocessing.Queue.get 方法？

问题描述

下面的代码将三个数字放在一个队列中。然后它尝试从队列中取回号码。但它永远不会。如何从队列中获取数据？

import multiprocessing

queue = multiprocessing.Queue()

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

while not queue.empty():
    print queue.get()

标签： pythonqueuemultiprocessingpython-multiprocessing

解决方案

在阅读@Martijn Pieters'之后，我最初删除了这个答案，因为他更详细和更早地描述了“为什么这不起作用”。然后我意识到，OP 示例中的用例不太适合

“如何使用 multiprocessing.Queue.get 方法”。

这并不是因为演示中不涉及子进程，而是因为在实际应用程序中，几乎不会预先填充队列并且仅在之后读取，但是读取和写入发生在中间的等待时间中交错发生。Martijn 展示的扩展演示代码在通常情况下不起作用，因为当排队跟不上读取速度时，while 循环会过早中断。所以这里是重新加载的答案，它能够处理通常的交错提要和读取场景：

不要依赖 queue.empty 检查同步。

将对象放入空队列后，队列的 empty() 方法返回 False 和 get_nowait() 可以在不引发 queue.Empty 的情况下返回之前可能会有一个无限小的延迟。...

空的（）

如果队列为空，则返回 True，否则返回 False。由于多线程/多处理语义，这是不可靠的。文档

要么从队列中使用for msg in iter(queue.get, sentinel):，.get()通过传递一个哨兵值来打破循环...... iter（callable，sentinel）？

from multiprocessing import Queue

SENTINEL = None

if __name__ == '__main__':

    queue = Queue()

    for i in [*range(3), SENTINEL]:
        queue.put(i)

    for msg in iter(queue.get, SENTINEL):
        print(msg)

...或者如果您需要非阻塞解决方案，请使用get_nowait()并处理可能的异常。queue.Empty

from multiprocessing import Queue
from queue import Empty
import time

SENTINEL = None

if __name__ == '__main__':

    queue = Queue()

    for i in [*range(3), SENTINEL]:
        queue.put(i)

    while True:
        try:
            msg = queue.get_nowait()
            if msg == SENTINEL:
                break
            print(msg)
        except Empty:
            # do other stuff
            time.sleep(0.1)

如果只有一个进程且该进程中只有一个线程正在读取队列，则还可以将最后一个代码片段交换为：

while True:
    if not queue.empty():  # this is not an atomic operation ...
        msg = queue.get()  # ... thread could be interrupted in between
        if msg == SENTINEL:
            break
        print(msg)
    else:
        # do other stuff
        time.sleep(0.1)

由于线程可以在检查和之间删除GIL，因此这不适用于进程中的多线程队列读取。如果多个进程正在从队列中读取，这同样适用。if not queue.empty()queue.get()

不过，对于单一生产者/单一消费者的场景，使用 amultiprocessing.Pipe而不是multiprocessing.Queue就足够了，而且性能更高。

推荐阅读