跳到内容

第 2 卷 / 第 3 章 / 第 2 课

3.2 队列:先进先出 ​

前置:数组与链表的选择、栈。这一节从 FIFO 契约出发,拆开循环索引的空满条件,再看队列如何保证 BFS 按层推进。

队列只认入队顺序 ​

先进先出(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 标准库通常这样写:

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:

text
next_index = (current_index + 1) % capacity

下面实现一个容量固定的循环队列。它用 _size 区分空与满,因此数组的 capacity 个位置全部可用。

python
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("队列检查通过")
text
入队后:[1, 2, 3, 4]
出队:1
绕回后:[2, 3, 4, 5]
队列已满:True
队列检查通过

手算一次绕回 ​

容量为 4,先入队 1、2、3、4:

时刻headtailsize逻辑顺序
初始000[]
依次入队 1..4004[1, 2, 3, 4]
出队 1103[2, 3, 4]
入队 5114[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 ​

本节代码采用这一套:

text
empty: size == 0
full:  size == capacity

head == tail 可能表示空,也可能表示满,要结合 size 判断。优点是全部槽位都可用,代价是多保存并维护一个计数。

方案 B:保留一个空槽 ​

另一种实现不存 size:

text
empty: head == tail
full:  (tail + 1) % storage_length == head

若想对外提供 N 个可用位置,底层通常要分配 N+1 个槽位。把这组判定抄进“记录 size”的实现,会造成容量少一格或空满混淆。

固定容量与动态扩容 ​

本节的 CircularQueue 在满时抛 OverflowError,适合演示有界缓冲区。动态循环队列可以在满时分配更大的数组,再按逻辑顺序复制:

python
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 让较早发现的层先得到处理。

下面用二叉树展示层序遍历:

python
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 个邻接项。在一条窄链上队列始终很短,所以不能说它对所有输入都必然平方退化。

python
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) 空间。

常见语言的队列接口 ​

语言常用队列入队 / 出队 / 查看失败与成本
Pythoncollections.dequeappend / popleft / q[0]两端操作近似 O(1);空弹出抛 IndexError
Java 17Queue<E> + ArrayDeque<E>offer / poll / peekpoll、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,而不是出队后才标记

动手验证绕回与容量 ​

  1. 容量 5 的循环队列从 head=4 开始有 3 个元素,它们的物理下标是什么?
  2. 把示例改成保留空槽方案。对外容量 4 时底层数组应有多长?
  3. 给动态循环队列设计扩容步骤。扩容后 head 和 tail 应是多少?
  4. 对一棵只有右孩子的树运行 BFS,最大队列长度是多少?
  5. 进阶:用两个栈实现 FIFO 队列,并用均摊分析解释转移成本。

最后核对 FIFO 不变量 ​

循环队列最容易错的不是取模公式,而是不变量混用:记录 size 时可以用满全部槽位;保留空槽时必须多分配一个位置。选定一种方案后,再把空、满、扩容和失败方式写进契约。BFS 依赖的也正是这份 FIFO 保证,先发现的节点先展开。

下一节讨论 双端队列:它在保留顺序契约的同时开放两端操作。

Built with VitePress | Software Systems Atlas