从线程队列获取所有项目

Question

我有一个线程将结果写入队列。

在另一个线程（GUI）中，我定期（在IDLE事件中）检查队列中是否有结果，如下所示：

def queue_get_all(q):
    items = []
    while 1:
        try:
            items.append(q.get_nowait())
        except Empty, e:
            break
    return items

这是一个很好的方法吗？

修改

  

我问，因为有时候   等待线程陷入困境   没有取出新的秒   结果

“卡住”问题原来是因为我在空闲事件处理程序中进行处理，而没有确保通过调用 wx.WakeUpIdle 实际生成这样的事件，这是推荐的。

Answer 1

如果 get_nowait（）调用导致暂停，如果列表为空则不返回，我会感到非常惊讶。

您是否在检查之间发布了大量（可能很大？）项目，这意味着接收线程有大量数据需要从 Queue 中提取出来？您可以尝试限制一批中检索的数字：

def queue_get_all(q):
    items = []
    maxItemsToRetreive = 10
    for numOfItemsRetrieved in range(0, maxItemsToRetreive):
        try:
            if numOfItemsRetrieved == maxItemsToRetreive:
                break
            items.append(q.get_nowait())
        except Empty, e:
            break
    return items

这会限制接收线程一次最多提取10个项目。

Answer 2

如果你总是将所有可用的项目从队列中拉出来，那么使用队列是否有任何实际意义，而不仅仅是一个带锁的列表？即：

from __future__ import with_statement
import threading

class ItemStore(object):
    def __init__(self):
        self.lock = threading.Lock()
        self.items = []

    def add(self, item):
        with self.lock:
            self.items.append(item)

    def getAll(self):
        with self.lock:
            items, self.items = self.items, []
        return items

如果您还单独提取它们，并利用空队列的阻塞行为，那么您应该使用Queue，但您的用例看起来更简单，并且可能通过上述方法更好地服务。

[Edit2] 我错过了你从空闲循环中轮询队列这一事实，而且从你的更新中，我发现问题与争用无关，所以以下方法与您的问题无关。如果有人发现这个有用的阻止变体，我就把它留下来了：

对于您希望在获得至少一个结果之前阻止的情况，您可以修改上述代码以等待数据通过生产者线程发出信号而变为可用。例如

class ItemStore(object):
    def __init__(self):
        self.cond = threading.Condition()
        self.items = []

    def add(self, item):
        with self.cond:
            self.items.append(item)
            self.cond.notify() # Wake 1 thread waiting on cond (if any)

    def getAll(self, blocking=False):
        with self.cond:
            # If blocking is true, always return at least 1 item
            while blocking and len(self.items) == 0:
                self.cond.wait()
            items, self.items = self.items, []
        return items

Answer 3

我认为将所有项目排除在队列之外的最简单方法如下：

def get_all_queue_result(queue):

    result_list = []
    while not queue.empty():
        result_list.append(queue.get())

    return result_list

Answer 4

我看到你正在使用get_nowait（）根据文档，“如果一个项目立即可用，则返回[s]项目，否则引发空例外”

现在，当抛出Empty异常时，您碰巧会跳出循环。因此，如果队列中没有立即可用的结果，则函数将返回空项列表。

您是否有理由不使用get（）方法？可能是get_nowait（）失败的情况，因为队列在同一时刻正在处理put（）请求。

Answer 5

如果您已完成对队列的写入，qsize应该可以完成这一操作，而无需为每次迭代检查队列。

responseList = []
for items in range(0, q.qsize()):
    responseList.append(q.get_nowait())

Answer 6

最简单的方法是使用列表解析：

items = [q.get() for _ in range(q.qsize())]

通常不赞成使用 range 函数，但我还没有找到更简单的方法。