3.2 队列:先进先出
队列只认入队顺序
先进先出(First In, First Out,FIFO)表示最早进入的元素最先离开。遵守这项契约的抽象数据类型叫队列;它不关心元素有多着急,插队要由另一套优先级合同负责。
队列把两端分开使用:
- 队尾(rear)接收新元素,操作叫
enqueue; - 队首(front)移除最早到达的元素,操作叫
dequeue。
栈在同一端压入和弹出,队列则从一端进入、另一端离开。二者的差别是公开契约,不是底层材料。数组、链表和双端队列都能实现 FIFO。
操作与失败契约
| 操作 | 含义 | 空队列 / 满队列时的约定 |
|---|---|---|
enqueue(x) | 把 x 加到队尾 | 有界队列已满时抛异常、阻塞或返回失败 |
dequeue() | 移除并返回队首 | 空队列时抛异常或返回空值 |
front() | 查看队首,不移除 | 空队列时抛异常或返回空值 |
is_empty() | 是否没有元素 | 空队列返回 True |
is_full() | 有界队列是否占满 | 无界队列通常不提供此语义 |
“有界”是接口的一部分。网络缓冲区可能固定最多容纳 N 项;普通应用中的动态队列则可能按需扩容。调用方必须知道满队列会丢弃、阻塞、失败,还是扩容。
为什么不把 Python list 的左端当队首
Python list.append(x) 在右端追加,连续操作的均摊成本是 Θ(1)。list.pop(0) 删除左端元素后,后面的引用要向前移动,因此一次操作是 Θ(n)。
list 当然能表达 FIFO,问题是左端删除的成本。队列规模始终很小时,这个差别可能不影响程序;规模会增长或操作频繁时,应使用端点操作高效的容器。
Python 标准库通常这样写:
from collections import deque
jobs = deque(["任务 A", "任务 B"])
jobs.append("任务 C")
first = jobs.popleft()
assert first == "任务 A"
assert list(jobs) == ["任务 B", "任务 C"]collections.deque 的两端追加和弹出具有近似 O(1) 的性能。它还支持更多双端操作,下一节会完整介绍。
循环队列:让数组位置重复使用
元素出队后,原数组位置应该能够再次使用。若数组队列每次出队都搬移元素,就浪费了这一点。循环队列保留两个索引:
head指向当前队首;tail指向下一次写入的位置。
索引到达数组末尾后,用取模回到 0:
next_index = (current_index + 1) % capacity下面实现一个容量固定的循环队列。它用 _size 区分空与满,因此数组的 capacity 个位置全部可用。
from operator import index as integer_index
class CircularQueue:
def __init__(self, capacity):
capacity = integer_index(capacity)
if capacity <= 0:
raise ValueError("capacity must be positive")
self._data = [None] * capacity
self._capacity = capacity
self._head = 0
self._tail = 0
self._size = 0
def enqueue(self, item):
if self.is_full():
raise OverflowError("enqueue to full queue")
self._data[self._tail] = item
self._tail = (self._tail + 1) % self._capacity
self._size += 1
def dequeue(self):
if self.is_empty():
raise IndexError("dequeue from empty queue")
value = self._data[self._head]
self._data[self._head] = None
self._head = (self._head + 1) % self._capacity
self._size -= 1
return value
def front(self):
if self.is_empty():
raise IndexError("front from empty queue")
return self._data[self._head]
def is_empty(self):
return self._size == 0
def is_full(self):
return self._size == self._capacity
def to_list(self):
return [
self._data[(self._head + offset) % self._capacity]
for offset in range(self._size)
]
def __len__(self):
return self._size
if __name__ == "__main__":
queue = CircularQueue(4)
for value in [1, 2, 3, 4]:
queue.enqueue(value)
print(f"入队后:{queue.to_list()}")
print(f"出队:{queue.dequeue()}")
queue.enqueue(5)
print(f"绕回后:{queue.to_list()}")
print(f"队列已满:{queue.is_full()}")
assert queue.front() == 2
assert len(queue) == 4
print("队列检查通过")入队后:[1, 2, 3, 4]
出队:1
绕回后:[2, 3, 4, 5]
队列已满:True
队列检查通过手算一次绕回
容量为 4,先入队 1、2、3、4:
| 时刻 | head | tail | size | 逻辑顺序 |
|---|---|---|---|---|
| 初始 | 0 | 0 | 0 | [] |
| 依次入队 1..4 | 0 | 0 | 4 | [1, 2, 3, 4] |
| 出队 1 | 1 | 0 | 3 | [2, 3, 4] |
| 入队 5 | 1 | 1 | 4 | [2, 3, 4, 5] |
tail 在数组末尾之后回到 0,把已经空出的槽位写成 5。物理数组可能是 [5, 2, 3, 4],逻辑队列仍从 head=1 开始读取,所以顺序是 [2, 3, 4, 5]。
to_list 只用于观察,它要访问全部元素,是 Θ(n)。enqueue、dequeue 和 front 在这份固定容量实现中都是 Θ(1)。创建时要初始化 capacity 个槽位,时间和存储空间为 Θ(capacity)。空满依据 _size,所以 None 也可以作为正常元素入队。
两种判空判满方案不要混用
循环队列常见两套不变量。
方案 A:记录 size
本节代码采用这一套:
empty: size == 0
full: size == capacityhead == tail 可能表示空,也可能表示满,要结合 size 判断。优点是全部槽位都可用,代价是多保存并维护一个计数。
方案 B:保留一个空槽
另一种实现不存 size:
empty: head == tail
full: (tail + 1) % storage_length == head若想对外提供 N 个可用位置,底层通常要分配 N+1 个槽位。把这组判定抄进“记录 size”的实现,会造成容量少一格或空满混淆。
固定容量与动态扩容
本节的 CircularQueue 在满时抛 OverflowError,适合演示有界缓冲区。动态循环队列可以在满时分配更大的数组,再按逻辑顺序复制:
def grow(old, head, size, new_capacity):
if not old or not 0 <= head < len(old) or not 0 <= size <= len(old):
raise ValueError("invalid circular buffer state")
if new_capacity <= len(old):
raise ValueError("new capacity must exceed old capacity")
new = [None] * new_capacity
for offset in range(size):
new[offset] = old[(head + offset) % len(old)]
return new这个辅助函数要求 head、size、new_capacity 都是整数,只返回新数组。调用者还要把 head 设为 0、tail 设为 size,并更新 capacity;否则下一次出队仍会按旧位置读取。
初始化新数组和复制分别需 Θ(new_capacity) 与 Θ(size) 工作。从空队列和固定初始容量开始,每次满时按固定倍数(大于 1)扩容,整段入队序列可得到均摊 Θ(1)。增长比例属于实现细节,不能从“动态队列”这个抽象推导出来。
BFS:谁先发现,谁先展开
广度优先搜索(BFS)先展开起点,再展开它的邻居,然后才到更远一层。队列保存“已经发现、尚未展开”的节点,FIFO 让较早发现的层先得到处理。
下面用二叉树展示层序遍历:
from collections import deque
class TreeNode:
def __init__(self, value, left=None, right=None):
self.value = value
self.left = left
self.right = right
def bfs_level_order(root):
if root is None:
return []
levels = []
queue = deque([root])
while queue:
level = []
for _ in range(len(queue)):
node = queue.popleft()
level.append(node.value)
if node.left is not None:
queue.append(node.left)
if node.right is not None:
queue.append(node.right)
levels.append(level)
return levels
root = TreeNode(
1,
TreeNode(2, TreeNode(4), TreeNode(5)),
TreeNode(3, None, TreeNode(6)),
)
assert bfs_level_order(root) == [[1], [2, 3], [4, 5, 6]]len(queue) 在每轮开始时记录当前层的节点数。循环只处理这一批;过程中加入的孩子留给下一轮。输入必须是真正的树,不含环或共享孩子。n 个节点各处理一次,时间 Θ(n);最大层宽为 w 时队列占 O(w),返回全部分层结果另占 Θ(n) 空间。
令 V、E 为起点可达部分的顶点数和邻接项数。若图使用邻接表、哈希查找和队列端点操作按常数时间计,BFS 的时间复杂度是 Θ(V+E):每个顶点至多按算法约定入队一次,每条边至多检查常数次。若改用 list.pop(0),每次出队都可能搬移当前队列里的引用;队列宽度能达到 Θ(V) 时,搬移成本在最坏情况下可到 Θ(V²),此外仍要检查 E 个邻接项。在一条窄链上队列始终很短,所以不能说它对所有输入都必然平方退化。
from collections import deque
def bfs_distances(graph, start):
distance = {start: 0}
queue = deque([start])
while queue:
node = queue.popleft()
for neighbor in graph[node]:
if neighbor not in distance:
distance[neighbor] = distance[node] + 1
queue.append(neighbor)
return distance
graph = {"A": ["B", "C"], "B": ["D"], "C": ["D"], "D": []}
assert bfs_distances(graph, "A") == {"A": 0, "B": 1, "C": 1, "D": 2}bfs_distances 要求起点和每个可达邻居都在字典中有邻接列表。distance 在入队前记录,兼做 visited,避免有环或汇合路径导致重复入队。返回值是无权图的最少边数,只含可达顶点;有权图不能直接用它比较总权重。队列和距离表各占 O(V) 空间。
常见语言的队列接口
| 语言 | 常用队列 | 入队 / 出队 / 查看 | 失败与成本 |
|---|---|---|---|
| Python | collections.deque | append / popleft / q[0] | 两端操作近似 O(1);空弹出抛 IndexError |
| Java 17 | Queue<E> + ArrayDeque<E> | offer / poll / peek | poll、peek 空时返回 null;remove、element 空时抛异常;ArrayDeque 禁止 null |
| C++ | std::queue<T> | push / pop / front | 默认适配 std::deque;先确认非空再读 front 或调用 pop,后者不返回元素 |
Java 的 Queue 接口允许非 FIFO 实现,例如优先队列;本表选用的 ArrayDeque 才按 FIFO 用法处理这些方法。ArrayDeque 是可增长数组实现,文档把大多数操作描述为均摊常数时间。它不是线程安全容器。多个步骤组成的“检查后执行”流程仍需外部同步或并发队列。
参考契约:
常见错误怎么定位
| 症状 | 先检查什么 |
|---|---|
| 入队后覆盖未出队元素 | 写入前是否检查满;tail 是否代表下一个写入位置 |
| 绕回后顺序错乱 | 访问位置是否使用 (head + offset) % capacity |
| 容量总少一格 | 是否把“保留空槽”判满规则混进 size 方案 |
| 出队后对象仍被持有 | 是否把旧 head 槽位清成 None |
| BFS 把两层混在一起 | 是否在加入孩子之前保存本层 len(queue) |
| 图 BFS 反复访问节点 | 是否在入队时标记 visited,而不是出队后才标记 |
动手验证绕回与容量
- 容量 5 的循环队列从
head=4开始有 3 个元素,它们的物理下标是什么? - 把示例改成保留空槽方案。对外容量 4 时底层数组应有多长?
- 给动态循环队列设计扩容步骤。扩容后
head和tail应是多少? - 对一棵只有右孩子的树运行 BFS,最大队列长度是多少?
- 进阶:用两个栈实现 FIFO 队列,并用均摊分析解释转移成本。
最后核对 FIFO 不变量
循环队列最容易错的不是取模公式,而是不变量混用:记录 size 时可以用满全部槽位;保留空槽时必须多分配一个位置。选定一种方案后,再把空、满、扩容和失败方式写进契约。BFS 依赖的也正是这份 FIFO 保证,先发现的节点先展开。
下一节讨论 双端队列:它在保留顺序契约的同时开放两端操作。