如何在Python中迭代Queue.Queue项目？

Question

有谁知道迭代 Queue.Queue 元素的 Pythonic 方法，而不将它们从队列中删除。我有一个生产者/消费者类型的程序，其中要处理的项目是使用 Queue.Queue 传递的，并且我希望能够打印剩余的项目是什么。有什么想法吗？

Answer 1

您可以循环访问底层数据存储的副本：

for elem in list(q.queue)

尽管这绕过了队列对象的锁，但列表副本是一个原子操作，应该可以正常工作。

如果您想保留锁，为什么不将所有任务从队列中取出，复制列表，然后将它们放回去。

mycopy = []
while True:
     try:
         elem = q.get(block=False)
     except Empty:
         break
     else:
         mycopy.append(elem)
for elem in mycopy:
    q.put(elem)
for elem in mycopy:
    # do something with the elements

Answer 2

列出队列元素而不消耗它们：

>>> from Queue import Queue
>>> q = Queue()
>>> q.put(1)
>>> q.put(2)
>>> q.put(3)
>>> print list(q.queue)
[1, 2, 3]

操作后，您仍然可以处理它们：

>>> q.get()
1
>>> print list(q.queue)
[2, 3]

Answer 3

您可以子类化 queue.Queue 以线程安全的方式实现此目的：

import queue


class ImprovedQueue(queue.Queue):
    def to_list(self):
        """
        Returns a copy of all items in the queue without removing them.
        """

        with self.mutex:
            return list(self.queue)

Answer 4

您可以在打印元素之前将双端队列转换为列表，以便可以轻松地迭代它。

from collections import deque

d = deque([7,9,3,5])

d.append(2)
d.appendleft(1)
d.append(10)
d.pop()

for elem in list(d):
    print(elem, end=" ")

#Output: 1 7 9 3 5 2 

Answer 5

我已经实现了一个支持async for迭代的IterableQueue(asyncio.Queue)。请参阅 pyutils 中的 GitHub。

from pyutils import IterableQueue
from asyncio import run, Task, create_task

async def producer(Q: IterableQueue[int], n: int) -> None:
    await Q.add_producer(N=1) 
    for i in range(n):
        await Q.put(i)
    await Q.finish()
    return None

async def amain():
    q : IterableQueue[int] = IterableQueue(maxsize=5)
    task : Task = create_task(producer(q, 10))
    # Iterate over queue items
    async for i in q:
        print(f"Got {i}")

if __name__ == "__main__":
    run(amain())

IterableQueue() 使用 add_ Producer() 对生产者进行计数。一旦最后一个生产者完成（finish()），那么一个哨兵值（None）就会添加到队列中，标记队列结束。