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

Question

有没有一种Pythonic的方法可以迭代一个Queue.Queue的元素 而不是 将它们从队列中移除。 我有一个生产者/消费者类型的程序，其中要处理的项通过使用Queue.Queue传递，我想能够打印剩余的项是什么。 有什么想法吗？

- pgilmon

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

- Raymond Hettinger

---

在Python 3中，对于队列（Queue）对象使用for elem in list(q.queue)会导致TypeError: 'Queue' object is not iterable的错误。也许是我的操作有误？ - Jonathan Komar

看起来你有另一层，其中“q”是一个代码对象，具有包含Queue对象的“queue”属性。尝试这样做：for elem in list(q.queue.queue)。 - Raymond Hettinger

明白了，会遵守。谢谢你的建议。（此消息将被删除） - Jonathan Komar

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]

- Omer Dagan

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)

- Erwin Mayer

Answer 4

你可以在打印元素之前将deque转换为list，这样你就可以轻松地遍历它。

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 

- Codemaker

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）。

- Jylpah