Python 队列

本文记录 Python 原生队列模块。

Python 队列

• queue 模块实现了多生产者、多消费者的队列。当必须在多个线程之间安全地交换信息时，它在线程编程中特别有用。该模块中的 Queue 类实现了所有必需的锁定语义。
• 该模块实现了三种类型的队列，它们仅在检索条目的顺序上有所不同。
  • **Queue **—— FIFO 队列: 最先添加的任务是最先检索。
  • LifoQueue——LIFO 队列: 最近添加的条目是第一个检索到的条目（像堆栈一样操作）。
  • PriorityQueue——优先级队列: 条目保持排序（使用 heapq 模块），并首先检索最低值的条目。
• 官方文档：https://docs.python.org/3/library/queue.html?highlight=queue#module-queue

队列类

Python 实现了 Queue，LifoQueue，PriorityQueue，SimpleQueue

• 所有的类
  • class queue.Queue(maxsize=0)
  • class queue.LifoQueue(maxsize=0)
  • class queue.PriorityQueue(maxsize=0)
  • class queue.SimpleQueue 简单队列没有上限限制，缺少任务跟踪等高级功能。

Queue

Queue 包含 Queue， LifoQueue， PriorityQueue 不包含 SimpleQueue

方法

• Queue.qsize() 返回队列的大致大小。注意，qsize() > 0 不保证后续的 get() 不会阻塞，qsize() < maxsize 也不保证 put() 不会阻塞。
• Queue.empty() 如果队列为空，则返回 True，否则返回 False。如果 empty() 返回 True，它并不能保证对 put() 的后续调用不会阻塞。类似地，如果 empty() 返回 False，它并不能保证对 get() 的后续调用不会阻塞。
• Queue.full() 如果队列已满，则返回 True，否则返回 False。如果 full() 返回 True，它并不能保证对 get() 的后续调用不会阻塞。类似地，如果 full() 返回 False，它并不能保证对 put() 的后续调用不会阻塞。
• Queue.put(item, block=True, timeout=None) 将 item 放入队列。如果可选参数 block 为 true 并且 timeout 为 None（默认值），则在必要时阻塞，直到有空闲插槽可用。如果 timeout 是一个正数，它最多会阻塞 timeout 秒，如果在该时间内没有可用的空闲槽，则会引发 Full 异常。否则（block 为假），如果空闲槽立即可用，则将项目放入队列，否则引发 Full 异常（在这种情况下忽略 timeout）。
• Queue.put_nowait(item) put(item, block=False) 的简写。
• Queue.get(block=True, timeout=None) 从队列中移除并返回一个项目。如果可选参数 block 为 true 并且 timeout 为 None（默认值），则在必要时阻塞，直到项目可用。如果 timeout 是一个正数，它最多会阻塞 timeout 秒，如果在那段时间内没有可用的项目，则会引发 Empty 异常。否则（block 为 false），如果一个项目立即可用，则返回一个项目，否则引发 Empty 异常（在这种情况下会忽略 timeout）。
• Queue.``get_nowait() get(False) 的简写。

以下两个方法被用来跟踪任务是否被守护进程消费者线程完全处理。

• Queue.task_done() 指示以前排队的任务已完成。由队列消费者线程使用。对于用于获取任务的每个 get()，对 task_done() 的后续调用会告诉队列该任务的处理已完成。如果 join() 当前处于阻塞状态，它将在处理完所有项目后恢复（这意味着对于已将 put() 放入队列的每个项目都收到了 task_done() 调用）。如果调用的次数多于队列中放置的项目，则引发 ValueError。
• Queue.join() 阻塞，直到队列中的所有项目都已被获取和处理。 每当将项目添加到队列中时，未完成任务的计数就会增加。每当消费者线程调用 task_done() 以指示该项目已被检索并且所有工作已完成时，计数就会下降。当未完成任务的计数降至零时，join() 会解除阻塞。

示例代码

import threading, queue

q = queue.Queue()

def worker():
    while True:
        item = q.get()
        print(f'Working on {item}')
        print(f'Finished {item}')
        q.task_done()

# Turn-on the worker thread.
threading.Thread(target=worker, daemon=True).start()

# Send thirty task requests to the worker.
for item in range(5):
    q.put(item)

# Block until all tasks are done.
q.join()
print('All work completed')

-->
Working on 0
Finished 0
Working on 1
Finished 1
Working on 2
Finished 2
Working on 3
Finished 3
Working on 4
Finished 4
All work completed

SimpleQueue

方法

• SimpleQueue.qsize() 如果队列为空，则返回 True，否则返回 False。如果 empty() 返回 False不能保证对 get() 的后续调用不会阻塞。
• SimpleQueue.put(item, block=True, timeout=None) 将项目放入队列。该方法从不阻塞并且总是成功（除了潜在的低级错误，例如未能分配内存）。可选的 args 块和超时被忽略，仅提供与 Queue.put() 的兼容性。
• SimpleQueue.put_nowait(item) 等效于 put(item, block=False)，提供与 Queue.put_nowait() 的兼容性。
• SimpleQueue.get(block=True, timeout=None) 从队列中移除并返回一个项目。如果可选的 args 块为真并且超时为无（默认值），则在必要时阻塞，直到项目可用。如果 timeout 是一个正数，它最多会阻塞 timeout 秒，如果在那段时间内没有可用的项目，则会引发 Empty 异常。否则（块为假），如果一个项目立即可用，则返回一个项目，否则引发空异常（在这种情况下忽略超时）。
• SimpleQueue.get_nowait() 等效于 get(False).

示例代码

import threading, queue

q = queue.SimpleQueue()


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

print(q.qsize())
while not q.empty():
    print(q.get())

-->
5
0
1
2
3
4

参考资料

• https://docs.python.org/3/library/queue.html?highlight=queue#module-queue