获取队列元素并在 Python 中迭代队列

目录

Get an item from a Queue in Python without removing it

1. 在 Python 中从队列中获取一个项目而不删除它
2. 在 Python 中获取队列的第一个元素
3. 检查元素是否在 Python 中的队列中
4. 在 Python 中将多个项目放入队列中
5. 在 Python 中遍历队列
6. 在 Python 中处理 queue.Empty 异常

在 Python 中从队列中获取一个项目而不删除它

使用queue队列上的属性从队列中获取项目而不删除它，例如q.queue[0]。

该queue属性指向一个deque支持索引的对象。

import queue

q = queue.Queue()

for item in range(10):
    q.put(item)


print(q.queue[0])  # 👉️ 0
print(q.queue[0])  # 👉️ 0

print(q.queue[1])  # 👉️ 1

If you don’t want to remove the specific element of the queue and only want to
access it, use q.queue[0].

This approach also works if you use the PriorityQueue class.

import queue

q = queue.PriorityQueue(maxsize=20)

for item in range(10):
    q.put(item)


print(q.queue[0])  # 👉️ 0
print(q.queue[0])  # 👉️ 0

You can also access the queue at a specific index if you use the
collections.deque
class.

from collections import deque

deq = deque(['a', 'b', 'c'])

print(deq[0])  # 👉️ 'a'
print(deq[1])  # 👉️ 'b'
print(deq[2])  # 👉️ 'c'

When we access the queue item at the specific index on a deque object, the
item remains in the queue.

Conversely, if you use the
get() method on
a queue object or the
popleft()
or
pop()
methods on a deque object, the item gets returned and removed from the queue.

# Get the First element of a Queue in Python

You can use the get() method to get the first element of a queue.

The get() method removes and returns an item from the queue.

import queue

q = queue.Queue()

for item in range(15):
    q.put(item)


print(q.queue[0])  # 👉️ 0 get first without removing it

# 👇️ remove and return an item from the queue
print(q.get())  # 👉️ 0 (get first)
print(q.get())  # 👉️ 1 (get second)
print(q.get())  # 👉️ 2 (get third)

If you don’t want to remove the element, use the queue attribute on the queue
and access the element at index 0.

If you use deque objects from the collections module, scroll down to the
next code snippet.

If you don’t want to remove the specific element of the queue and only want to
access it, use q.queue[0].

The queue.get
method removes and returns an item from the queue.

If you use the collections.deque class, you can access the first element in
the queue by accessing the deque object at index 0.

from collections import deque

deq = deque(['a', 'b', 'c'])

print(deq[0])  # 👉️ get first
print(deq[1])  # 👉️ get second
print(deq[2])  # 👉️ get third

first = deq.popleft()
print(first)  # 👉️ 'a'

print(deq) # 👉️ deque(['b', 'c'])

Deque objects also support the
popleft()
method.

The method removes and returns an element from the left side of the deque.

If no elements are present in the deque, the method raises an IndexError.

If you need to remove and return an element from the right side of the deque,
use the
pop()
method.

from collections import deque

deq = deque(['a', 'b', 'c'])

print(deq[0])  # 👉️ get first

first = deq.popleft()
print(first)  # 👉️ 'a'

print(deq)  # 👉️ deque(['b', 'c'])

last = deq.pop()
print(last) # 👉️ 'c'

print(deq)  # 👉️ deque(['b'])

# Check if an element is in a Queue in Python

If you need to check if an element is in a queue:

1. Access the queue attribute on the queue to get a deque object.
2. Use the in operator to check if the element is in the queue.
3. The in operator tests for membership.

import queue

q = queue.Queue()

q.put(0)
q.put(1)

if 0 in q.queue:
    # 👇️ this runs
    print('0 is in queue')

if 100 not in q.queue:
    # 👇️ this runs
    print('100 is NOT in queue')

The queue attribute on the queue returns a deque object. Deque objects
support indexing and membership testing.

The in operator tests
for membership. For example, x in q evaluates to True if x is a member of
q, otherwise it evaluates to False.

If you used the
collections.deque
class to initialize a deque object, you can directly use the in operator to
check if an item is in the deque.

from collections import deque

deq = deque(['a', 'b'])

if 'a' in deq:
    print('a is in deque')

if 'z' not in deq:
    print('z is NOT in deque')

The collections.deque class has atomic append(), implements the popleft()
method and supports indexing and membership testing.

If you used the queue module, access the queue attribute on the queue to get
a deque object.

import queue

q = queue.Queue()

q.put(0)
q.put(1)

print(q.queue) # 👉️ deque([0, 1])

# Put multiple items in a Queue in Python

If you need to put multiple items in a queue:

1. Use a for loop to iterate over an iterable or a range.
2. On each iteration call the put() method on the iterable.
3. Pass the item to the put() method.

import queue

q = queue.Queue()

my_list = list(range(10))
print(my_list)

# 👇️ iterate over list and put each item in the queue
for item in my_list:
    q.put(item)

print(q.queue)  # 👉️ deque([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

for item in q.queue:
    print(item)  # 👉️ 0 1 2 3 4 ...

We used a for loop to iterate over a list
containing 10 items and used the put() method to put each item in the queue.

The Queue.put
method puts an item in the queue.

You can also use a range if you need to put N items in the queue.

import queue

q = queue.Queue()

# 👇️ iterate over range and put each item in the queue
for item in range(10):
    q.put(item)

print(q.queue)  # 👉️ deque([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

for item in q.queue:
    print(item)  # 👉️ 0 1 2 3 4 ...

If you use a deque object, you can pass an iterable directly to the extend()
method.

from collections import deque

deq = deque()

my_list = list(range(10))
deq.extend(my_list)

print(deq)  # 👉️ deque([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

for item in deq:
    print(item)  # 0 1 2 3 4 ...

We used the
collections.deque
class to create a deque object.

The
extend
method takes an iterable as an argument and extends the right side of the deque
by appending the items from the iterable.

You can also use a for loop with the append() method.

from collections import deque

deq = deque()

print(deq)

my_list = list(range(10))

for item in my_list:
    deq.append(item)

print(deq)  # 👉️ deque([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])

for item in deq:
    print(item)  # 👉️ 0 1 2 3 4 ...

The
append
method takes a value and adds it to the right side of the deque.

# Iterate through a Queue in Python

You can use a while loop to iterate through a queue.

import queue

q = queue.Queue()

for item in range(10):
    q.put(item)

while not q.empty():
    item = q.get()
    print(item)  # 👉️ 0 1 2 3 4 ...

The loop checks if the queue is not empty and iterates as long as there are
items in the queue.

The
queue.empty
method returns True if the queue is empty, and False otherwise.

We use this approach to prevent blocking after the queue has been emptied.

Note that the
queue.get method
removes and returns an item from the queue.

Alternatively, you can use the queue attribute on the queue to get access to a
deque object and iterate over the object using a for loop.

import queue

q = queue.Queue()

for item in range(10):
    q.put(item)


for item in q.queue:
    print(item) # 👉️ 0 1 2 3 4 ...

Deque objects support indexing and iteration, so we can iterate over a queue as
if we were iterating over a list.

Alternatively, you can use the iter() function.

import queue

q = queue.Queue()

for item in range(10):
    q.put(item)


sentinel = object()

for item in iter(q.get, sentinel):
    print(item) # 👉️ 0 1 2 3 4 ...

    # 👇️ prevent queue from blocking after tasks finished
    q.put(sentinel) # 👈️ do this last

The iter() function returns an
iterator object.

After all of the work is finished, we add a sentinel value to the queue to raise
StopIteration and prevent from blocking.

# Handle the queue.Empty exception in Python

Use a
try/except statement to
handle the queue.Empty exception.

import queue

q = queue.Queue()

q.get(False)

try:
    item = q.get(False)

    # 👉️ do work here

    # 👇️ indicate task complete
    q.task_done()
except queue.Empty:
    print('queue.Empty exception')
    # 👉️ handle empty queue exception
    pass

You can use the get() method in the try block and catch the queue.Empty
exception in the except block.

The Queue.get
method removes and returns an item from the queue.

We passed False to the method to make it non-blocking.

You can do all the necessary work in the try block and call the
Queue.task_done
method at the end to indicate that the task is complete.

我们可以使用该task_done()方法在每次调用该get()方法以获取任务后告诉队列任务已完成。

使用else块

您还可以通过使用块来稍微简化一下else。

import queue

q = queue.Queue()

try:
    item = q.get(False)

    # 👉️ do work here

except queue.Empty:
    print('queue.Empty exception')
    # handle empty queue exception
    pass
else:
    # 👇️ indicate task complete
    q.task_done()

该try/except语句有一个可选else子句。

该else块仅在try块不引发异常时运行。

我还写了一篇关于
如何获取 Queue 长度的文章。