一、队列概述
队列（Queue）是一种先进先出（FIFO, First-In-First-Out）的线性数据结构。它的工作方式类似于现实生活中的排队——先到的人先获得服务，后到的人排在队尾等待。
队列的核心特性：
- FIFO原则： 最先加入队列的元素最先被移除
- 受限访问： 只能在队尾（rear）添加元素，在队首（front）移除元素
- 基本操作： enqueue（入队）、dequeue（出队）、peek（查看队首）、is_empty（判空）、size（获取长度）
- 时间复杂度： 理想情况下所有操作均为 O(1)
队列的基本操作定义
一个完整的队列接口包含以下核心方法：
enqueue(item) — 将元素添加到队尾dequeue() — 移除并返回队首元素peek() / front() — 返回队首元素但不移除is_empty() — 检查队列是否为空size() — 返回队列中元素的数量
"队列是一种受限制的线性表，限定只在表的一端进行插入操作、在另一端进行删除操作。允许插入的一端称为队尾（rear），允许删除的一端称为队首（front）。"
二、普通队列的实现
2.1 基于列表（数组）的实现
Python 中最简单的队列实现方式是使用内置的 list。使用 append() 实现入队操作（O(1)均摊），使用 pop(0) 实现出队操作（O(n)，因为需要移动所有元素）。
class ListQueue:
def __init__(self):
self.items = []
def enqueue(self, item):
"""入队：在列表末尾添加元素，O(1)均摊"""
self.items.append(item)
def dequeue(self):
"""出队：移除并返回列表首部元素，O(n)"""
if self.is_empty():
raise IndexError("队列为空")
return self.items.pop(0)
def peek(self):
"""查看队首元素"""
if self.is_empty():
raise IndexError("队列为空")
return self.items[0]
def is_empty(self):
return len(self.items) == 0
def size(self):
return len(self.items)
def __str__(self):
return "Queue: front -> " + " -> ".join(str(x) for x in self.items) + " -> rear"
性能分析： 这种实现简单直观，但 dequeue() 操作的时间复杂度为 O(n)，因为 pop(0) 需要将列表中的所有元素向前移动一位。对于大量数据，这会成为性能瓶颈。
2.2 基于链表（Linked List）的实现
使用双向链表可以实现 enqueue 和 dequeue 均为 O(1) 的队列。利用链表头结点作为队首、尾结点作为队尾。
class Node:
def __init__(self, value):
self.value = value
self.next = None
class LinkedListQueue:
def __init__(self):
self.front = None # 队首指针
self.rear = None # 队尾指针
self._size = 0
def enqueue(self, item):
"""入队：在链表尾部添加，O(1)"""
node = Node(item)
if self.rear is None:
self.front = self.rear = node
else:
self.rear.next = node
self.rear = node
self._size += 1
def dequeue(self):
"""出队：移除链表头部节点，O(1)"""
if self.is_empty():
raise IndexError("队列为空")
value = self.front.value
self.front = self.front.next
if self.front is None:
self.rear = None
self._size -= 1
return value
def peek(self):
if self.is_empty():
raise IndexError("队列为空")
return self.front.value
def is_empty(self):
return self._size == 0
def size(self):
return self._size
性能分析： 链表实现下，入队和出队操作均为 O(1)。但每个节点需要额外存储指针（内存开销较大），且缓存局部性不如数组实现。
三、循环队列（Circular Queue）
循环队列是数组实现队列的一种优化方式。通过复用已出队元素的存储空间，避免了 pop(0) 带来的 O(n) 数据搬移开销。循环队列使用 head 指针（指向队首）和 tail 指针（指向队尾下一个位置）来管理队列。
循环队列核心原理
- head 指针： 指向队首元素的位置
- tail 指针： 指向下一个入队元素应该存放的位置
- 判空条件：
head == tail
- 判满条件：
(tail + 1) % capacity == head（牺牲一个存储单元以区分空和满）
- 指针移动： 使用模运算
(pointer + 1) % capacity 实现循环
class CircularQueue:
def __init__(self, capacity=10):
self.items = [None] * (capacity + 1) # 多开一个位置区分空/满
self.capacity = capacity + 1
self.head = 0 # 队首指针
self.tail = 0 # 队尾指针（指向下一个插入位置）
def enqueue(self, item):
"""入队：O(1)"""
if self.is_full():
raise IndexError("队列已满")
self.items[self.tail] = item
self.tail = (self.tail + 1) % self.capacity
def dequeue(self):
"""出队：O(1)"""
if self.is_empty():
raise IndexError("队列为空")
value = self.items[self.head]
self.items[self.head] = None
self.head = (self.head + 1) % self.capacity
return value
def peek(self):
if self.is_empty():
raise IndexError("队列为空")
return self.items[self.head]
def is_empty(self):
return self.head == self.tail
def is_full(self):
return (self.tail + 1) % self.capacity == self.head
def size(self):
return (self.tail - self.head + self.capacity) % self.capacity
循环队列设计要点：
- 利用
(tail + 1) % capacity == head 判断队列是否已满，此时数组中有一个空位未使用
size() 的计算公式为 (tail - head + capacity) % capacity，确保模运算结果不为负- 所有操作均为 O(1)，没有数据搬移开销
- 适用于有界队列场景，如操作系统 I/O 缓冲区、网络数据包缓冲
使用示例：
# 循环队列使用演示
cq = CircularQueue(5)
for i in range(5):
cq.enqueue(i * 10)
print(cq.dequeue()) # 输出: 0
print(cq.dequeue()) # 输出: 10
cq.enqueue(99) # 复用已释放的空间
print(cq.peek()) # 输出: 20
print(cq.size()) # 输出: 4
四、双端队列（Deque）
双端队列（Deque, Double-Ended Queue）是一种可以在两端进行插入和删除操作的线性数据结构。与普通队列不同，Deque 没有严格的 FIFO 限制，允许在队首和队尾两端进行 O(1) 的添加和移除操作。
4.1 Python 的 collections.deque
Python 标准库提供了 collections.deque，它是用双向链表（准确地说，是双向的块状链表）实现的，支持两端的 O(1) 操作。
from collections import deque
# 创建双端队列
d = deque(['a', 'b', 'c'])
# 两端添加元素
d.append('d') # 右端添加: deque(['a','b','c','d'])
d.appendleft('x') # 左端添加: deque(['x','a','b','c','d'])
# 两端弹出元素
right = d.pop() # 右端弹出: 'd', deque变为['x','a','b','c']
left = d.popleft() # 左端弹出: 'x', deque变为['a','b','c']
# 其他常用操作
d.extend(['e', 'f']) # 右端扩展
d.extendleft(['m', 'n']) # 左端扩展（注意顺序反转）
d.rotate(1) # 向右循环旋转1步
d.rotate(-2) # 向左循环旋转2步
print(d[0]) # O(1) 访问首部
print(d[-1]) # O(1) 访问尾部
collections.deque vs list 性能对比
| 操作 | deque | list |
|---|
| 右端添加（append） | O(1) | O(1)均摊 |
| 右端弹出（pop） | O(1) | O(1) |
| 左端添加（appendleft） | O(1) | O(n) |
| 左端弹出（popleft） | O(1) | O(n) |
| 索引访问 | O(n) | O(1) |
| 中间插入 | O(n) | O(n) |
4.2 用 deque 实现栈和队列
利用 deque 可以轻松实现栈（Stack）和队列（Queue）两种数据结构：
from collections import deque
class Stack:
"""基于 deque 实现栈（后进先出）"""
def __init__(self):
self._data = deque()
def push(self, item):
self._data.append(item)
def pop(self):
return self._data.pop()
def peek(self):
return self._data[-1]
def is_empty(self):
return len(self._data) == 0
class Queue:
"""基于 deque 实现队列（先进先出）"""
def __init__(self):
self._data = deque()
def enqueue(self, item):
self._data.append(item)
def dequeue(self):
return self._data.popleft()
def peek(self):
return self._data[0]
def is_empty(self):
return len(self._data) == 0
五、优先队列（PriorityQueue）
优先队列（Priority Queue）是一种特殊的队列，每个元素都有一个优先级属性。出队操作总是返回当前队列中优先级最高的元素（或最低的，取决于实现）。优先队列的底层实现通常是二叉堆（Binary Heap）。
5.1 二叉堆（Binary Heap）原理
二叉堆的关键性质
- 完全二叉树： 除最后一层外，其他层节点是满的，最后一层节点靠左排列
- 堆序性： 最小堆中每个父节点的值 <= 所有子节点的值；最大堆中每个父节点的值 >= 所有子节点的值
- 数组存储： 一般使用数组实现，根节点在索引0，任意节点 i 的左子节点在
2*i+1，右子节点在 2*i+2，父节点在 (i-1)//2
- 核心操作：
heapify（建堆，O(n)）、push（插入，O(log n)）、pop（取出最值，O(log n)）
5.2 Python heapq 模块
Python 标准库的 heapq 模块提供了最小堆（Min-Heap）的实现。所有操作基于列表进行原地调整。
import heapq
# 创建堆
heap = []
numbers = [5, 3, 8, 1, 6, 9, 2, 7, 4]
for num in numbers:
heapq.heappush(heap, num)
# 每步插入后，堆保持最小堆性质
# 弹出最小值
smallest = heapq.heappop(heap) # 返回 1
# 查看最小值（不弹出）
print(heap[0]) # 堆顶元素即最小值
# 从列表直接构建堆 — O(n)
arr = [9, 7, 5, 3, 1, 8, 6, 4, 2]
heapq.heapify(arr) # 原地建堆
# 同时 push 和 pop
result = heapq.heappushpop(heap, 0) # 先 push 0，再 pop 最小值
result = heapq.heapreplace(heap, 10) # 先 pop 最小值，再 push 10
# 获取最大的 N 个元素 / 最小的 N 个元素
largest = heapq.nlargest(3, [1, 5, 3, 9, 7, 2]) # [9, 7, 5]
smallest = heapq.nsmallest(3, [1, 5, 3, 9, 7, 2]) # [1, 2, 3]
5.3 自定义优先级
当需要根据复杂规则确定优先级时，可以在元组中组合优先级值，或者实现自定义类的 __lt__ 方法。
import heapq
from dataclasses import dataclass, field
# 方式一：使用元组（优先级, 数据）
tasks = []
heapq.heappush(tasks, (3, "写报告"))
heapq.heappush(tasks, (1, "处理紧急bug"))
heapq.heappush(tasks, (2, "代码审查"))
priority, task = heapq.heappop(tasks)
print(task) # 输出: "处理紧急bug"（优先级最高，数字最小）
# 方式二：自定义对象实现 __lt__
@dataclass
class Patient:
name: str
severity: int # 病情严重程度（数字越大越严重）
arrival_order: int = field(compare=False)
def __lt__(self, other):
# 病情更严重的优先级更高（最大堆效果）
if self.severity != other.severity:
return self.severity > other.severity
# 同等病情时，先到的优先级高
return self.arrival_order < other.arrival_order
# 模拟急诊室分诊
pq = []
heapq.heappush(pq, Patient("张三", 5, 1))
heapq.heappush(pq, Patient("李四", 9, 2)) # 更严重，先处理
heapq.heappush(pq, Patient("王五", 5, 3))
while pq:
p = heapq.heappop(pq)
print(f"处理患者: {p.name}, 严重程度: {p.severity}")
# 输出顺序: 李四(9) -> 张三(5) -> 王五(5)
5.4 使用 queue.PriorityQueue
Python 标准库也提供了线程安全的 queue.PriorityQueue 类，其底层也是基于 heapq 实现的。
from queue import PriorityQueue
pq = PriorityQueue()
pq.put((2, "中等优先级"))
pq.put((1, "高优先级"))
pq.put((3, "低优先级"))
while not pq.empty():
priority, item = pq.get()
print(f"优先级 {priority}: {item}")
# 输出:
# 优先级 1: 高优先级
# 优先级 2: 中等优先级
# 优先级 3: 低优先级
heapq vs PriorityQueue 对比
- heapq： 非线程安全，性能更高，适合单线程场景
- queue.PriorityQueue： 线程安全（内部使用锁），适合多线程场景
- queue.PriorityQueue： 提供了 put()/get() 方法的阻塞机制，支持超时
- queue.PriorityQueue： 适合生产者-消费者模式
六、队列的典型应用
6.1 广度优先搜索（BFS）
队列是 BFS 算法的核心数据结构。无论在图还是树中，BFS 都使用队列来逐层遍历节点。
from collections import deque
def bfs_graph(graph, start):
"""
图的广度优先遍历
graph: 邻接表表示的图 {node: [neighbors]}
start: 起始节点
"""
visited = set()
queue = deque([start])
visited.add(start)
result = []
while queue:
node = queue.popleft()
result.append(node)
for neighbor in graph[node]:
if neighbor not in visited:
visited.add(neighbor)
queue.append(neighbor)
return result
# 示例：二叉树层序遍历
class TreeNode:
def __init__(self, val=0):
self.val = val
self.left = None
self.right = None
def level_order(root):
"""二叉树层序遍历（BFS）"""
if not root:
return []
result = []
queue = deque([root])
while queue:
level_size = len(queue)
level = []
for _ in range(level_size):
node = queue.popleft()
level.append(node.val)
if node.left:
queue.append(node.left)
if node.right:
queue.append(node.right)
result.append(level)
return result
6.2 任务调度（Task Scheduling）
操作系统中的任务调度广泛使用队列。普通队列用于轮询调度（Round Robin），优先队列用于优先级调度。
import heapq
import time
class TaskScheduler:
"""基于优先级的任务调度器"""
def __init__(self):
self._queue = [] # 优先队列
self._counter = 0 # 相同优先级时保证 FIFO
def add_task(self, name, priority=0):
"""添加任务，priority 越小优先级越高"""
heapq.heappush(self._queue, (priority, self._counter, name))
self._counter += 1
def run_next(self):
"""执行下一个优先级最高的任务"""
if not self._queue:
return
_, _, name = heapq.heappop(self._queue)
print(f"[{time.strftime('%H:%M:%S')}] 执行: {name}")
def run_all(self):
"""按优先级执行所有任务"""
while self._queue:
self.run_next()
# 模拟 CPU 进程调度
scheduler = TaskScheduler()
scheduler.add_task("系统时钟中断", priority=1)
scheduler.add_task("用户程序 - 浏览器", priority=5)
scheduler.add_task("磁盘 I/O 完成", priority=2)
scheduler.add_task("用户程序 - 编辑器", priority=5)
scheduler.add_task("网络数据包到达", priority=1)
scheduler.run_all()
# 执行顺序: 系统时钟中断 -> 网络数据包到达 -> 磁盘 I/O 完成 -> 浏览器 -> 编辑器
6.3 生产者-消费者模式
队列是解决生产者-消费者问题的经典方案，queue.Queue 提供了线程安全的实现。
from queue import Queue
import threading
import time
import random
def producer(q, name):
"""生产者：向队列中放入数据"""
for i in range(5):
item = f"{name}-产品-{i}"
q.put(item)
print(f"[生产者 {name}] 生产: {item}")
time.sleep(random.uniform(0.1, 0.5))
def consumer(q, name):
"""消费者：从队列中取出数据并处理"""
while True:
item = q.get()
print(f"[消费者 {name}] 消费: {item}")
time.sleep(random.uniform(0.2, 0.6))
q.task_done()
# 创建线程安全队列，最大容量为 10
q = Queue(maxsize=10)
# 启动 2 个生产者和 3 个消费者
for i in range(2):
t = threading.Thread(target=producer, args=(q, chr(65 + i)))
t.daemon = True
t.start()
for i in range(3):
t = threading.Thread(target=consumer, args=(q, chr(88 + i)))
t.daemon = True
t.start()
# 等待所有生产者完成
print("主线程等待所有任务完成...")
q.join()
print("所有任务处理完毕！")
# queue.Queue 的关键特性：
# - put(): 如果队列已满则阻塞，直到有空位
# - get(): 如果队列为空则阻塞，直到有数据
# - task_done(): 通知队列一个任务已完成
# - join(): 阻塞直到所有任务都被处理完毕
6.4 消息队列（Message Queue）
在分布式系统和微服务架构中，消息队列是核心组件。虽然生产环境通常使用 RabbitMQ、Kafka、Redis 等专业中间件，但其核心原理仍然是队列数据结构。
消息队列的典型使用场景
- 异步处理： 将耗时操作（如发送邮件、生成报表）放入消息队列，快速返回响应
- 流量削峰： 高并发请求先进入消息队列，后端服务按处理能力消费，防止系统过载
- 系统解耦： 生产者和消费者不需要直接通信，通过消息队列作为中间层
- 日志收集： 多个服务将日志写入消息队列，日志处理系统集中消费
from queue import Queue
import threading
import time
# 简单的异步任务处理系统（模拟消息队列）
task_queue = Queue()
# 后台工作线程
def worker():
while True:
task = task_queue.get()
print(f"[Worker] 处理任务: {task}")
time.sleep(1) # 模拟任务处理时间
task_queue.task_done()
threading.Thread(target=worker, daemon=True).start()
# 模拟 Web 请求提交异步任务
def handle_request(req_id):
print(f"[Web] 收到请求 #{req_id}，已加入队列")
task_queue.put(f"请求 #{req_id} 的处理任务")
for i in range(5):
handle_request(i)
task_queue.join()
print("所有请求处理完成")
七、时间复杂度分析
下表总结了各队列实现的核心操作时间复杂度：
| 实现方式 | enqueue / push | dequeue / pop |
peek | space |
| list（列表模拟） |
O(1) 均摊 |
O(n) |
O(1) |
O(n) |
| 链表实现 |
O(1) |
O(1) |
O(1) |
O(n) * |
| 循环队列 |
O(1) |
O(1) |
O(1) |
O(n) |
| collections.deque |
O(1) |
O(1) |
O(1) |
O(n) * |
| heapq（优先队列） |
O(log n) |
O(log n) |
O(1) |
O(n) |
| queue.PriorityQueue |
O(log n) |
O(log n) |
— |
O(n) |
* 链表和 deque 的额外空间开销来自指针/节点元数据。
选择建议：
- 普通 FIFO 队列： 首选
collections.deque，两端操作均为 O(1)
- 有界队列 / 缓冲区： 使用循环队列，固定内存占用，无动态扩容开销
- 优先级调度： 使用
heapq（单线程）或 queue.PriorityQueue（多线程）
- 线程间通信： 使用
queue.Queue（线程安全，支持阻塞）
- 算法题 / 竞赛： 使用
collections.deque，性能优越且 API 丰富
八、核心要点总结
- 队列的核心原则： 先进先出（FIFO）是队列的根本特性，与之对应的是栈的先进后出（LIFO）
- 三种主流实现： 数组（循环队列）、链表、deque（双向块状链表），各有优劣
- 循环队列关键： head/tail 指针配合模运算实现 O(1) 操作；通过牺牲一个存储单元来区分空和满
- 双端队列（Deque）： 两端均可 O(1) 插入/删除，可同时作为栈和队列使用
- 优先队列底层： 二叉堆（完全二叉树 + 堆序性），建堆 O(n)、插入/删除 O(log n)
- Python heapq： 最小堆实现，支持 heappush、heappop、heapify、heapreplace、nlargest/nsmallest
- BFS 与队列： 队列是广度优先搜索的天然搭档，逐层遍历节点
- 生产者-消费者： queue.Queue 提供线程安全的阻塞队列，是并发编程的基础组件
- 时间复杂度黄金标准： 普通队列操作应为 O(1)，优先队列操作应为 O(log n)
九、进一步思考
队列作为一种基础数据结构，其思想已经渗透到计算机科学的各个领域。深入理解队列及其变体，不仅是掌握数据结构的需要，更是理解系统设计的起点。
扩展方向
- 阻塞队列（Blocking Queue）： Java 的 ArrayBlockingQueue、LinkedBlockingQueue，Python 的 queue.Queue 都是阻塞队列的实现，常用于线程池和连接池
- 延迟队列（Delay Queue）： 元素在指定延迟时间后才能被取出，常用于定时任务和过期缓存清理
- 并发无锁队列（Lock-Free Queue）： 使用 CAS（Compare-And-Swap）指令实现的高性能并发队列，如 Python 的 multiprocessing.Queue
- 单调队列（Monotonic Queue）： 队列中的元素保持单调递增或递减，常用于解决滑动窗口最值问题（如 LeetCode 239 滑动窗口最大值）
- 双端队列与滑动窗口： 结合双端队列可以 O(n) 时间内解决滑动窗口最大值/最小值问题
LeetCode 经典题目推荐
- 普通队列： 225. 用队列实现栈、232. 用栈实现队列
- 双端队列： 239. 滑动窗口最大值、862. 和至少为 K 的最短子数组
- 优先队列： 23. 合并 K 个升序链表、215. 数组中的第 K 个最大元素、347. 前 K 个高频元素
- BFS： 102. 二叉树的层序遍历、200. 岛屿数量、752. 打开转盘锁
- 生产者-消费者： 1188. 设计有限阻塞队列（多线程）
实际工程中的应用
- Web 服务器： Nginx 的事件队列、请求队列使用队列管理连接
- 数据库： MySQL 的 redo log buffer、binlog 传输使用队列机制
- 操作系统： CPU 调度队列（就绪队列、等待队列）、I/O 请求队列
- 网络： 数据包缓冲队列、TCP 的滑动窗口机制
- 异步框架： Python asyncio 的事件循环使用队列管理协程调度
- 分布式系统： Kafka 的分区日志本质上是一个持久化的队列
"队列不仅仅是数据结构课上的一种抽象数据类型。理解队列的设计哲学——解耦、缓冲、异步——对于构建健壮的软件系统至关重要。FIFO 的原则看似简单，但当它与优先级、阻塞、超时、持久化等特性结合时，就演化出了现代系统中不可或缺的中间件基础设施。"