queue线程安全队列：Queue/LifoQueue/PriorityQueue

Python并发编程专题 · 最安全高效的线程间数据交换方式

专题：Python并发编程系统学习

关键词：Python, 并发编程, queue, Queue, LifoQueue, PriorityQueue, 线程安全队列

一、为什么需要线程安全队列

共享内存的竞态条件问题

在多线程编程中，多个线程共享同一进程的内存空间。当多个线程同时对同一个共享变量进行读写操作时，就会产生竞态条件（Race Condition）。例如，两个线程同时执行 counter += 1，这个操作在底层实际上分三步执行：读取counter值、加1、写回counter。如果两个线程的读取操作交替执行，最终counter的值可能只增加了1而不是2。

传统的解决方式是使用锁（Lock）来保护共享数据，但手动管理锁存在诸多问题：忘记释放锁导致死锁、锁的粒度控制不当导致性能下降、复杂的锁嵌套逻辑极难调试。

队列封装了锁机制

queue.Queue 模块在内部已经封装了 threading.Lock（用于保证 put 和 get 操作的互斥）和 threading.Condition（用于实现阻塞和唤醒机制）。开发者只需要调用 put() 和 get() 方法，无需关心底层的锁操作，就能实现线程安全的数据交换。

Queue vs 手动锁

使用 Queue 比手动加锁有以下优势：第一，更简洁——两行代码即可完成安全的存取操作；第二，更安全——避免了忘记释放锁导致的死锁；第三，更强大——内置阻塞和超时机制，天然支持生产者-消费者模式；第四，更灵活——可以方便地切换不同队列类型（FIFO/LIFO/优先级）。

核心要点：queue.Queue 是 Python 线程安全数据交换的首选方案。除非有特殊需求，否则优先使用 Queue 而不是手动管理 Lock + list。Queue 在内部已经实现了所有必要的同步机制，让开发者可以专注于业务逻辑。

二、Queue（FIFO先进先出）

Queue 是标准的先进先出（FIFO）队列。第一个放入的元素会第一个被取出，就像超市排队结账一样。这是最常用的队列类型，适用于任务调度、消息传递、数据流处理等绝大多数生产者-消费者场景。

核心方法

put(item, block=True, timeout=None)：将元素放入队列。如果队列已满，block=True 时阻塞直到有空位；timeout 设置最长等待时间（超时抛出 queue.Full 异常）。
get(block=True, timeout=None)：从队列取出元素。如果队列为空，block=True 时阻塞直到有元素；timeout 设置最长等待时间（超时抛出 queue.Empty 异常）。
task_done()：通知队列一个任务已经处理完毕。每当消费者取走元素并完成处理后调用，用于配合 join() 计数。
join()：阻塞直到队列中所有元素都被处理完毕（即每个 put 的元素都被对应的 task_done 调用计数）。
qsize()：返回队列当前的大致元素数量（注意：在多线程环境下这个值并不精确，仅供估算）。
empty() / full()：判断队列是否为空/已满（同样是近似判断，不保证绝对精确）。

经典生产者-消费者模式

import queue
import threading
import time
import random

q = queue.Queue(maxsize=10)

def producer(name):
    for i in range(20):
        item = f"{name}-{i}"
        q.put(item)
        print(f"[生产者 {name}] 放入: {item}, 队列大小: {q.qsize()}")
        time.sleep(random.uniform(0.1, 0.5))
    q.join()  # 等待所有task_done被调用
    print(f"[生产者 {name}] 所有任务处理完毕")

def consumer(name):
    while True:
        item = q.get()
        print(f"[消费者 {name}] 取出: {item}")
        time.sleep(random.uniform(0.2, 0.8))
        q.task_done()  # 标记任务完成

# 启动生产者和消费者
p = threading.Thread(target=producer, args=("A",))
c = threading.Thread(target=consumer, args=("X",), daemon=True)

p.start()
c.start()
p.join()  # 等待生产者完成

put/get 阻塞行为详解

当队列已满时调用 put()，如果 block=True（默认值），线程会阻塞直到队列有空位。这天然实现了背压（backpressure）机制——生产者不会无限制地生产，而是被消费者处理速度所调节。反之，当队列为空时 get() 会阻塞，消费者线程会等待生产者放入新数据。这种双向阻塞机制是实现稳健的生产者-消费者模式的基础。

import queue

q = queue.Queue(maxsize=3)

# 非阻塞 put - 队列满时抛出异常
try:
    q.put("A", block=False)
    q.put("B", block=False)
    q.put("C", block=False)
    q.put("D", block=False)  # 队列已满，抛出 queue.Full
except queue.Full:
    print("队列已满，无法放入")

# 带超时的 put
try:
    q.put("D", timeout=2)  # 最多等待2秒
except queue.Full:
    print("等待2秒后队列仍然满")

三、LifoQueue（后进先出）

LifoQueue 是后进先出（LIFO）队列，行为类似于栈（Stack）。最后放入的元素会最先被取出。这种结构适用于深度优先搜索（DFS）、撤销操作、最近优先处理等场景。

基本用法

LifoQueue 拥有与 Queue 完全相同的 API——put()、get()、task_done()、join()、qsize() 等方法完全一致。唯一的区别是内部数据结构的存取顺序：LifoQueue 使用 list 模拟栈的行为，后放入的元素优先被取出。

import queue

lq = queue.LifoQueue(maxsize=10)
lq.put("第一个任务")
lq.put("第二个任务")
lq.put("第三个任务")

# 输出顺序: 第三个任务 -> 第二个任务 -> 第一个任务
while not lq.empty():
    try:
        item = lq.get(timeout=1)
        print(f"取出: {item}")
    except queue.Empty:
        break

应用场景：深度优先遍历

当需要遍历树或图结构且希望优先深入某个分支时，LifoQueue 是理想的选择。每次将当前节点的所有子节点放入 LifoQueue，然后取出最顶层的节点继续处理，天然实现了深度优先的遍历逻辑。

import queue

def dfs_tree(root):
    """使用LifoQueue实现树的深度优先遍历"""
    lq = queue.LifoQueue()
    lq.put(root)
    visited = []

    while True:
        try:
            node = lq.get(timeout=1)
        except queue.Empty:
            break
        if node not in visited:
            visited.append(node)
            print(f"访问节点: {node.value}")
            # 将子节点逆序放入，以保证从左到右访问
            for child in reversed(node.children):
                lq.put(child)
    return visited

在实际开发中，如果只是需要简单的栈结构（不涉及多线程），建议使用普通的 list 作为栈。LifoQueue 的优势在于它提供了线程安全的操作，适合多个线程同时进行 push/pop 的场景。

四、PriorityQueue（优先队列）

PriorityQueue 是优先队列，内部使用小顶堆（heapq）实现。放入的元素以元组 (priority, data) 的形式存储，priority 值越小的元素越先被取出。这在任务调度、事件驱动系统、图算法（如 Dijkstra 最短路径）中非常实用。

基本用法

import queue

pq = queue.PriorityQueue()

pq.put((3, "低优先级任务"))
pq.put((1, "高优先级任务"))
pq.put((2, "中优先级任务"))

while not pq.empty():
    priority, task = pq.get()
    print(f"优先级={priority}, 任务={task}")

# 输出:
# 优先级=1, 任务=高优先级任务
# 优先级=2, 任务=中优先级任务
# 优先级=3, 任务=低优先级任务

优先级规则详解

PriorityQueue 使用 heapq 模块实现。当元组的第一个元素（priority）相同时，会比较第二个元素。如果第二个元素是不可比较的类型，会抛出 TypeError。解决方法是给每个元组额外添加一个唯一标识符来打破平局：

import queue
import itertools

pq = queue.PriorityQueue()
counter = itertools.count()  # 生成唯一序号

# 使用(priority, counter, data)格式避免优先级相同时的比较问题
pq.put((2, next(counter), "A"))
pq.put((1, next(counter), "B"))
pq.put((2, next(counter), "C"))

# 当priority相同时，按counter（插入顺序）排序
while not pq.empty():
    _, _, task = pq.get()
    print(task)

应用场景：任务调度

import queue
import threading
import time
import random

class TaskScheduler:
    def __init__(self):
        self.task_queue = queue.PriorityQueue()
        self.counter = itertools.count()

    def add_task(self, priority, func, *args):
        self.task_queue.put((priority, next(self.counter), func, args))

    def _worker(self):
        while True:
            try:
                _, _, func, args = self.task_queue.get(timeout=1)
                print(f"执行任务: {func.__name__}, 参数={args}")
                func(*args)
                self.task_queue.task_done()
            except queue.Empty:
                break

    def run(self, num_workers=3):
        workers = []
        for _ in range(num_workers):
            t = threading.Thread(target=self._worker, daemon=True)
            t.start()
            workers.append(t)
        self.task_queue.join()  # 等待所有任务完成

scheduler = TaskScheduler()
scheduler.add_task(3, print, "低优先级的消息")
scheduler.add_task(1, print, "紧急消息！")
scheduler.add_task(2, print, "普通消息")
scheduler.run()

五、队列的阻塞与超时

put/get 的 block 和 timeout 机制

所有三种队列类型的 put() 和 get() 方法都接受 block 和 timeout 两个可选参数，这是队列实现线程同步的核心机制。

import queue

q = queue.Queue(maxsize=2)

# ----- put 的三种模式 -----
# 1. 阻塞模式（默认）：队列满时阻塞
q.put("A")
q.put("B")
# q.put("C")  # 这行会被阻塞，直到有空位

# 2. 非阻塞模式：队列满时立即抛出 queue.Full
try:
    q.put("C", block=False)
except queue.Full:
    print("非阻塞put: 队列已满")

# 3. 超时模式：最多等待N秒，超时抛出 queue.Full
try:
    q.put("C", timeout=2)
except queue.Full:
    print("超时put: 等待2秒后队列仍然满")

# ----- get 的三种模式 -----
# 清空队列
q.get()
q.get()

# 1. 阻塞模式（默认）：队列空时阻塞
# q.get()  # 这行会被阻塞

# 2. 非阻塞模式
try:
    q.get(block=False)
except queue.Empty:
    print("非阻塞get: 队列为空")

# 3. 超时模式
try:
    q.get(timeout=3)
except queue.Empty:
    print("超时get: 等待3秒后队列仍然空")

empty() 和 qsize() 的局限性

需要特别注意：Queue.empty() 和 Queue.qsize() 在多线程环境中是不精确的。当你调用 empty() 返回 False（队列非空）后，另一个线程可能瞬间取走了最后一个元素。同样，qsize() 返回的值在当前线程获取到返回值时可能已经过时。因此，不要依赖这两个方法做精确控制，应该依靠 put()/get() 的阻塞和异常机制来处理边界情况。

最佳实践：不要使用 while not q.empty(): item = q.get() 这样的模式。正确的做法是使用 try: item = q.get(timeout=N) 结合 except queue.Empty。这样可以避免在 empty() 和 get() 之间的时间窗口中其他线程修改了队列状态导致的竞态条件。

六、队列高级使用技巧

多消费者多生产者模式

Queue 天然支持多个生产者和多个消费者同时操作。内部锁机制确保了所有 put/get 操作都是线程安全的，不需要额外的同步措施。

import queue
import threading
import time
import random

q = queue.Queue()

def producer(name):
    for i in range(10):
        item = f"{name}-{i}"
        q.put(item)
        time.sleep(random.uniform(0.05, 0.15))

def consumer(name, results):
    processed = 0
    while True:
        try:
            item = q.get(timeout=2)
            print(f"[{name}] 处理: {item}")
            processed += 1
            q.task_done()
        except queue.Empty:
            break
    results[name] = processed

# 启动多个生产者和消费者
threads = []
results = {}
for p_name in ["P1", "P2", "P3"]:
    t = threading.Thread(target=producer, args=(p_name,))
    threads.append(t)
    t.start()

for c_name in ["C1", "C2", "C3", "C4"]:
    t = threading.Thread(target=consumer, args=(c_name, results))
    threads.append(t)
    t.start()

# 等待所有生产者完成
for t in threads[:3]:
    t.join()

# 等待队列中的所有元素被处理
q.join()

# 发送哨兵退出信号的另一种方式：
# 实际项目中常用 sentinel 模式，在所有任务后放入特殊标记

print("结果统计:", results)

join 与 task_done 的协作机制

Queue 内部维护了一个计数器。每次调用 put() 时计数器加1，每次调用 task_done() 时计数器减1。join() 方法阻塞直到这个计数器归零。这是实现"等待所有任务完成"的标准方式。注意：task_done() 必须由消费者在真正完成处理后调用，且每个 get 都应该对应一个 task_done，否则 join() 将永远阻塞。

重要：调用了 get() 但没有调用 task_done() 会导致 join() 永久阻塞。join() 的等待是基于 put 的次数和 task_done 的次数的匹配，和 get 本身无关。在编写消费者循环时，务必确保每个 get() 都有对应的 task_done()，即使处理过程发生异常也不例外（使用 try/finally）。

def safe_consumer():
    """使用try/finally确保task_done被调用"""
    while True:
        item = q.get()
        try:
            # 处理任务（可能抛出异常）
            process(item)
        finally:
            q.task_done()  # 无论如何都要调用

队列容量规划

创建队列时可以指定 maxsize 参数来限制最大容量。合理设置 maxsize 可以防止生产者过度生产导致内存膨胀。如果 maxsize 为 0 或负数（默认值），队列大小不受限制。在流式处理场景中，建议设置合理的 maxsize 以形成背压，使生产者的生产速率与消费者的消费速率达成平衡。

七、queue vs 其他同步方式对比

Python 提供了多种线程间数据交换和同步的方式，以下是 Queue 与其他方式的详细对比：

同步方式	线程安全	阻塞支持	使用复杂度	适用场景
queue.Queue	是（内置锁）	是（put/get阻塞）	低（两行代码）	数据交换、任务调度、生产者-消费者
Lock + list	需手动加锁	无（需额外实现）	高（易出错）	极简场景、不需要阻塞的队列
threading.Condition	需配合Lock	是（wait/notify）	高（需精确控制）	复杂的同步逻辑、多条件等待
Pipe（multiprocessing）	是（进程安全）	部分支持	中	进程间双向通信
multiprocessing.Queue	是（进程安全）	是	低	多进程数据交换
asyncio.Queue	是（协程安全）	是（await put/get）	低	异步协程中的数据流

选择建议：在 threading 场景下，优先使用 queue.Queue；在多进程场景下使用 multiprocessing.Queue；在 asyncio 场景下使用 asyncio.Queue。只有在需要对同步逻辑进行精细控制的特殊情况下，才考虑手动使用 Lock + Condition + 自定义数据结构。

八、核心要点总结

Queue 三大类型：

Queue — 先进先出（FIFO），适用于大多数生产者-消费者场景
LifoQueue — 后进先出（LIFO），适用于深度优先遍历和最近优先处理
PriorityQueue — 优先级队列，适用于任务调度和事件排序

核心API：put() / get() 实现线程安全的存取；task_done() / join() 实现任务完成等待；block / timeout 参数控制阻塞行为。

最佳实践：使用 try/except 处理 queue.Empty 和 queue.Full 异常而非依赖 empty()/full() 判断；使用 try/finally 确保 task_done() 被调用；合理设置 maxsize 实现背压。

九、进一步思考

Queue 虽然简单易用，但理解其内部实现有助于更深入地掌握并发编程。Queue 的核心是使用 Condition 对象的 wait() 和 notify() 方法来实现 put 和 get 的阻塞与唤醒。每个队列内部维持了一个 mutex 锁和两个 Condition（not_full 和 not_empty）。put 操作先获取锁，如果队列满了就在 not_full 上等待；放入数据后在 not_empty 上通知等待的 get 操作。

在实际项目中，建议根据具体需求选择合适的队列类型。如果只是需要线程间的安全数据传递，标准 Queue 已经足够。如果涉及任务优先级，PriorityQueue 是更好的选择。而如果需要"后进先出"的访问模式，LifoQueue 可以很好地满足需求。掌握这三种队列类型及其使用技巧，能够有效提升并发编程的代码质量和开发效率。