Python的多进程中如何检查空队列

Question

我有一个使用Python的multiprocessing和Queue包的程序。我的一个函数具有以下结构：

from multiprocessing import Process, Queue
def foo(queue):
   while True:
       try:
           a = queue.get(block = False)
           doAndPrintStuff(a)
       except:
           print "the end"
           break

   if __name__ == "__main__"
     nthreads = 4
     queue = Queue.Queue()
     # put stuff in the queue here 
     for stuff in moreStuff:
         queue.put(stuff)
     procs = [Process(target = foo, args = (queue,)) for i in xrange(nthreads)]
     for p in procs:
       p.start()
     for p in procs:
       p.join()

“这个想法是，当我尝试从队列中提取数据时，如果队列为空，它会引发一个异常并终止循环。所以我有两个问题：

1）这是一种安全的习惯用法吗？有更好的方法吗？

2）我试图找出在尝试从空队列中获取数据时引发的确切异常是什么。目前，我的程序捕获所有异常，当错误发生在其他地方时，只能得到一个“结束”的消息，这很糟糕。

我尝试了：

”

  import Queue
  queue = Queue.Queue()
  [queue.put(x) for x in xrange(10)]
  try: 
       print queue.get(block = False)
  except Queue.Empty:
       print "end"
       break

但我收到了错误，就好像我没有捕获异常一样。应该捕获什么样的异常才是正确的？

- Rafael S. Calsaverini

Answer 1

例外情况应该是Queue.Empty。但您确定收到了相同的错误吗？在您的第二个示例中，您还将队列本身从multiprocessing.Queue切换到Queue.Queue，我认为这可能是问题所在。

这可能看起来很奇怪，但您必须使用multiprocessing.Queue类，但使用Queue模块中自己导入的Queue.Empty异常。

- Steven

---

哦！谢谢！我不知道我需要从不同的包中导入东西。现在它可以工作了。 - Rafael S. Calsaverini

Answer 2

看起来队列在刷新缓冲区之前是空的，这可能需要一段时间。

我们问题的解决方案是使用sentinels或者使用内置的task_done()调用：

task_done()

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

如果join()当前正在阻塞，则当所有项目都被处理（意味着对队列中放置的每个项目都收到了task_done()调用）时，它将恢复。

如果调用次数多于放置在队列中的项目数量，则引发ValueError。

- Cees Timmerman

Answer 3

这里有一个例子- 正如@Steven在上面所说，您需要使用标准队列中的queue.Empty异常。来自文档的注释 (https://docs.python.org/3/library/multiprocessing.html):

注意

multiprocessing使用通常的queue.Empty和queue.Full异常来表示超时。它们在multiprocessing命名空间中不可用，因此您需要从queue中导入它们。

基本示例:

 from multiprocessing import Process, Queue, Manager
 import queue

def firstPass(q):
    driver = getDriver()

    while True:
        try:      
            link = q.get_nowait()   
            f(driver, link)
        except queue.Empty:
            logger.info("empty queue")
            driver.close()
            break

- grantr

Answer 4

试着阅读queue库的文档。你是不是在找Queue.empty()?

- Ross Patterson