Python 库手册：queue 线程安全队列模块|python|queue|spiderlinebreakprint|线程|队列

Python 的 queue 模块提供了线程安全的队列类，是构建多线程程序时处理任务调度、生产者-消费者模型的理想选择。它支持先进先出（FIFO）、后进先出（LIFO）以及优先级队列等结构，确保多个线程之间安全地共享数据。

常见应用场景：

（1）多线程环境中任务调度与线程通信。

（2）构建生产者-消费者模型。

（3）需要线程间共享数据但避免使用显式锁。

（4）需要带有优先级的任务队列

◆ ◆ ◆

核心概念

1、线程安全（Thread-safe）

所有队列操作都是原子级，内部使用锁机制，无需额外加锁。

2、三种队列类型

Queue：先进先出队列（FIFO）

LifoQueue：后进先出队列（栈结构）

PriorityQueue：按优先级出队，较小值优先

3、队列满/空时阻塞与非阻塞

入队和出队操作可选择是否阻塞，默认阻塞直到成功。

4、可设置最大队列长度（maxsize）

防止无限增长导致内存溢出。

◆ ◆ ◆

应用举例

例 1：基础 FIFO 队列操作

import queue

q = queue.Queue(maxsize=3)  # 创建容量为3的队列

q.put(10)
q.put(20)
q.put(30)

print(q.get())  # 输出 10
print(q.get())  # 输出 20

例 2：非阻塞取出元素

import queue

q = queue.Queue()
q.put(1)

try:
    item = q.get_nowait()  # 非阻塞读取
    print(item)
except queue.Empty:
    print("队列为空")

例 3：使用 LifoQueue 实现栈结构

import queue

s = queue.LifoQueue()
s.put('A')
s.put('B')
s.put('C')

print(s.get())  # 输出 C（后进先出）

例 4：使用 PriorityQueue 实现优先级任务调度

import queue

pq = queue.PriorityQueue()
pq.put((2, "中等优先级"))
pq.put((1, "高优先级"))
pq.put((3, "低优先级"))

while not pq.empty():
    print(pq.get()[1])
# 输出顺序：高优先级 -> 中等优先级 -> 低优先级

例 5：多线程中的生产者-消费者模型

import threading
import queue
import time
import random

# 创建一个共享队列
task_queue = queue.Queue(maxsize=5)

# 生产者函数
def producer(name):
    for i in range(10):
        item = f"{name}-任务{i}"
        task_queue.put(item)
        print(f"[生产者{name}] 生产了 {item}")
        time.sleep(random.uniform(0.1, 0.5))  # 模拟生产时间

# 消费者函数
def consumer(name):
    while True:
        try:
            item = task_queue.get(timeout=2)  # 等待任务
            print(f"    [消费者{name}] 消费了 {item}")
            task_queue.task_done()  # 标记任务完成
            time.sleep(random.uniform(0.2, 0.6))  # 模拟处理时间
        except queue.Empty:
            break

# 创建线程
producer_thread = threading.Thread(target=producer, args=("A",))
consumer_threads = [
    threading.Thread(target=consumer, args=("X",)),
    threading.Thread(target=consumer, args=("Y",))
]

# 启动线程
producer_thread.start()
for t in consumer_threads:
    t.start()

# 等待所有线程结束
producer_thread.join()
task_queue.join()  # 等待所有任务完成
print("所有任务处理完成")

◆ ◆ ◆

常用方法与属性

Queue(maxsize=0)

创建一个先进先出的线程安全队列。

参数：

maxsize：最大容量，0 表示无限制

返回：一个 Queue 对象

.put(item, block=True, timeout=None)

将一个元素放入队列。

参数：

item：要加入队列的元素

block：是否阻塞等待，默认 True