Python queue 模块
在 Python 中,queue 模块提供了一个线程安全的队列实现,用于在多线程编程中安全地传递数据。
队列是一种先进先出(FIFO)的数据结构,queue 模块提供了多种队列类型,包括 Queue、LifoQueue 和 PriorityQueue,以满足不同的需求。
队列类型
1. Queue
Queue 是 queue 模块中最常用的队列类型,它实现了标准的先进先出(FIFO)队列。以下是 Queue 的基本用法:
实例
# 创建一个队列
q = queue.Queue()
# 向队列中添加元素
q.put(1)
q.put(2)
q.put(3)
# 从队列中获取元素
print(q.get()) # 输出: 1
print(q.get()) # 输出: 2
print(q.get()) # 输出: 3
2. LifoQueue
LifoQueue 是一种后进先出(LIFO)的队列,类似于栈。以下是 LifoQueue 的基本用法:
实例
# 创建一个 LIFO 队列
q = queue.LifoQueue()
# 向队列中添加元素
q.put(1)
q.put(2)
q.put(3)
# 从队列中获取元素
print(q.get()) # 输出: 3
print(q.get()) # 输出: 2
print(q.get()) # 输出: 1
3. PriorityQueue
PriorityQueue 是一种优先级队列,元素按照优先级顺序被取出。以下是 PriorityQueue 的基本用法:
实例
# 创建一个优先级队列
q = queue.PriorityQueue()
# 向队列中添加元素,元素为元组 (优先级, 数据)
q.put((3, 'Low priority'))
q.put((1, 'High priority'))
q.put((2, 'Medium priority'))
# 从队列中获取元素
print(q.get()) # 输出: (1, 'High priority')
print(q.get()) # 输出: (2, 'Medium priority')
print(q.get()) # 输出: (3, 'Low priority')
常用方法
1. put(item, block=True, timeout=None)
将 item 放入队列。如果 block 为 True 且队列已满,则等待 timeout 秒,直到队列有空闲空间。如果 timeout 为 None,则无限等待。
2. get(block=True, timeout=None)
从队列中获取并移除一个元素。如果 block 为 True 且队列为空,则等待 timeout 秒,直到队列中有元素。如果 timeout 为 None,则无限等待。
3. qsize()
返回队列中的元素数量。
4. empty()
如果队列为空,返回 True,否则返回 False。
5. full()
如果队列已满,返回 True,否则返回 False。
线程安全
queue 模块的所有队列类型都是线程安全的,这意味着多个线程可以安全地同时操作同一个队列,而不需要额外的同步机制。这使得 queue 模块成为多线程编程中传递数据的理想选择。
示例:多线程队列
以下是一个使用 Queue 在多线程之间传递数据的示例:
实例
import threading
import time
# 创建一个队列
q = queue.Queue()
# 生产者线程
def producer():
for i in range(5):
print(f'生产 {i}')
q.put(i)
time.sleep(1)
# 消费者线程
def consumer():
while True:
item = q.get()
if item is None:
break
print(f'消费 {item}')
q.task_done()
# 启动生产者线程
producer_thread = threading.Thread(target=producer)
producer_thread.start()
# 启动消费者线程
consumer_thread = threading.Thread(target=consumer)
consumer_thread.start()
# 等待生产者线程完成
producer_thread.join()
# 等待队列中的所有任务完成
q.join()
# 发送结束信号
q.put(None)
consumer_thread.join()
常用的属性和方法
以下是 Python queue 模块(线程安全队列)的常用类、方法及属性的表格说明,包含功能描述和示例:
queue 模块核心类
| 类 | 说明 | 适用场景 |
|---|---|---|
queue.Queue | 先进先出(FIFO)队列 | 通用任务队列 |
queue.LifoQueue | 后进先出(LIFO)队列(类似栈) | 需要后进先出的场景 |
queue.PriorityQueue | 优先级队列(最小堆实现) | 按优先级处理任务 |
queue.SimpleQueue | 更简单的FIFO队列(Python 3.7+) | 不需要高级功能的场景 |
通用方法(所有队列类都支持)
| 方法 | 说明 | 示例 | 返回值 |
|---|---|---|---|
put(item) | 放入元素 | q.put("task1") | None |
get() | 取出并移除元素 | item = q.get() | 队列元素 |
empty() | 判断队列是否为空 | if q.empty(): | True/False |
full() | 判断队列是否已满 | if q.full(): | True/False |
qsize() | 返回队列当前大小 | size = q.qsize() | 整数 |
task_done() | 标记任务完成(用于join()) | q.task_done() | None |
join() | 阻塞直到所有任务完成 | q.join() | None |
阻塞控制参数
| 参数 | 说明 | 默认值 | 示例 |
|---|---|---|---|
block | 当队列为空/满时是否阻塞 | True | q.get(block=False) |
timeout | 阻塞超时时间(秒) | None | q.put(x, timeout=5) |
PriorityQueue 专用用法
元素格式:(priority, data),优先级越小越先出队
实例
pq.put((1, "low"))
pq.put((0, "high"))
print(pq.get()[1]) # 输出: "high"
实例
生产者-消费者模型:
实例
q = queue.Queue(maxsize=3) # 容量为3的队列
def producer():
for i in range(5):
q.put(f"Task-{i}")
print(f"Produced: Task-{i}")
def consumer():
while True:
item = q.get()
print(f"Consumed: {item}")
q.task_done()
threading.Thread(target=producer, daemon=True).start()
threading.Thread(target=consumer, daemon=True).start()
q.join() # 等待所有任务完成
优先级任务处理:
实例
pq.put((3, "Scan"))
pq.put((1, "Emergency"))
pq.put((2, "Log"))
while not pq.empty():
print(pq.get()[1]) # 输出顺序: Emergency → Log → Scan
非阻塞获取(避免死锁):
实例
item = q.get_nowait() # 等价于 q.get(block=False)
except queue.Empty:
print("队列为空")