尽管队列已满，但 Python 多处理队列 get() 超时

Question

我正在使用 Python 的多处理模块来进行科学并行处理。在我的代码中，我使用了几个执行繁重工作的工作进程和一个将结果保存到磁盘的写入器进程。要写入的数据通过队列从工作进程发送到写入进程。数据本身相当简单，仅由一个包含文件名的元组和一个带有两个浮点数的列表组成。经过几个小时的处理，编写器进程通常会卡住。更准确地说是下面的代码块

while (True):
    try:
        item = queue.get(timeout=60)
        break
    except Exception as error:
        logging.info("Writer: Timeout occurred {}".format(str(error)))

永远不会退出循环，我会收到连续的“超时”消息。

我还实现了一个日志记录过程，它输出队列的状态等，即使我收到上面的超时错误消息，对 qsize() 的调用也会不断返回一个完整的队列（在我的情况下大小 = 48 )。

我已彻底检查了有关队列对象的文档，但找不到可能的解释为什么 get() 在队列满时返回超时。

有什么想法吗？

编辑：

我修改了代码以确保捕获到空队列异常：

while (True):
    try:
        item = queue.get(timeout=60)
        break
    except Empty as error:
        logging.info("Writer: Timeout occurred {}".format(str(error)))

Answer 1

在多处理队列中用作同步消息队列。您的问题似乎也是这种情况。然而，这不仅仅需要调用get() 方法。处理完每个任务后，您需要调用 task_done() 以便从队列中删除该元素。

来自文档：

Queue.task_done()

表示以前排队的任务已完成。由队列消费者线程使用。 对于用于获取任务的每个 get()，对 task_done() 的后续调用会告诉队列该任务的处理已完成。

如果 join() 当前处于阻塞状态，它将在处理完所有项目后恢复（意味着对于已将 put() 放入队列的每个项目都收到了 task_done() 调用）。

在文档中，您还可以找到正确使用线程队列的代码示例。

如果你的代码应该是这样的

while (True):
    try:
        item = queue.get(timeout=60)
        if item is None:
            break
        # call working fuction here
        queue.task_done()
    except Exception as error:
        logging.info("Writer: Timeout occurred {}".format(str(error)))

Answer 2

切换到基于管理器的队列应该有助于解决此问题。

manager = Manager()
queue   = manager.Queue()

有关更多详细信息，您可以在此处查看多处理文档：https://docs.python.org/2/library/multiprocessing.html

Answer 3

您发现了一个过于笼统的Exception 并假设它是一个超时错误。

尝试修改逻辑如下：

from Queue import Empty

while (True):
    try:
        item = queue.get(timeout=60)
        break
    except Empty as error:
        logging.info("Writer: Timeout occurred {}".format(str(error)))
        print(queue.qsize())

并查看日志是否仍然打印。