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

Question

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

我需要知道队列不为空，以便我可以将队列的峰值(不删除它)与我刚刚使用 get 删除的元素进行比较

如何通过 sleep 来完成:

while myQueue.empty():
    sleep(0.05)

如何在不使用 sleep 的情况下做到这一点？我应该使用 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_empty条件，但我们不再需要 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]