如何在不浪费太多cpu周期的情况下等待多线程队列不为空

Question

我想让一个线程等到多线程队列不为空。队列只有一个生产者和一个消费者。生产者在任务可用时将任务放入队列，但生产者必须等到收集到两个或更多任务。我不只是使用该get方法两次来检索两个任务的原因是因为它使算法流程过于复杂。但是，这不能在下面的代码段中描述，因为显然这只是一个过于简单的例子。

我需要知道队列不是空的，以便我可以将队列的峰值（不删除它）与我刚刚删除的元素进行比较 get

如何通过睡眠来完成：

while myQueue.empty():
    sleep(0.05)

如何在不使用睡眠的情况下做到这一点？我应该使用event.wait()吗？如果是，我无法弄清楚应该如何正确使用该event.clear()命令。由于我想要等待的线程也是消费者，我无法确定队列是否为空。即使我queue.empty()用来检查。

Answer 1

从本质上讲，您似乎需要实现该Queue.peek()方法，该方法将返回队列中的下一个元素而不实际删除它。

此方法在标准 Queue 对象中不可用，但您可以毫无问题地继承和扩展它：

from Queue import Queue
class VoyeurQueue(Queue):
    def peek(self, block=True, timeout=None):
        # ...

现在对于新peek()方法的内容，您可以简单地复制粘贴get()基础Queue对象的方法内容并进行一些修改。/usr/lib/python?.?/Queue.py如果你在 Linux 上，或者%PYTHONPATH%/lib/Queue.py如果你在 Windows上，你可以找到它（不确定后者，因为我目前在 Linux 机器上并且无法检查）。在我的 Python 2.7 副本中，该get()方法实现为：

def get(self, block=True, timeout=None):
    # ... lots of comments
    self.not_empty.acquire()
    try:
        if not block:
            if not self._qsize():
                raise Empty
        elif timeout is None:
            while not self._qsize():
                self.not_empty.wait()
        elif timeout < 0:
            raise ValueError("'timeout' must be a non-negative number")
        else:
            endtime = _time() + timeout
            while not self._qsize():
                remaining = endtime - _time()
                if remaining <= 0.0:
                    raise Empty
                self.not_empty.wait(remaining)
        item = self._get()
        self.not_full.notify()
        return item
    finally:
        self.not_empty.release()

def _get(self):
    return self.queue.popleft()

现在，对于差异。您不想删除元素，因此_get()我们定义以下内容：

def _peek(self):
    return self.queue[0]

在peek()方法中，我们仍然使用self.not_emptyCondition 但我们不再需要self.not_full.notify(). 因此生成的代码将如下所示：

from Queue import Queue

class VoyeurQueue(Queue):

    def peek(self, block=True, timeout=None):
        self.not_empty.acquire()
        try:
            if not block:
                if not self._qsize():
                    raise Empty
            elif timeout is None:
                while not self._qsize():
                    self.not_empty.wait()
            elif timeout < 0:
                raise ValueError("'timeout' must be a non-negative number")
            else:
                endtime = _time() + timeout
                while not self._qsize():
                    remaining = endtime - _time()
                    if remaining <= 0.0:
                        raise Empty
                    self.not_empty.wait(remaining)
            item = self._peek()
            return item
        finally:
            self.not_empty.release()

    def _peek(self):
        return self.queue[0]