7.2 优先队列
上一节手写了堆。工程代码通常更关心接口：怎样加入任务，怎样查看下一项，优先级相同时谁先出来，取消任务后会发生什么。
优先队列是抽象数据类型，堆是它的一种实现。 未排序数组也能实现优先队列，尾部加入可做摊销 O(1)，但每次出队要扫描 O(n)；有序数组能让出队便宜，却把插入推到 O(n)。二叉堆的上滤与下滤都至多为 O(log n)，查看根是 O(1)。
堆的数组关系没有改变：父节点 (i - 1) // 2，左孩子 2 * i + 1，右孩子 2 * i + 2。优先队列在这层结构之上定义业务顺序。
1. 同优先级需要第二把尺子
假设调度器同时收到两个优先级为 2 的字典任务。Python 若只存 (priority, task)，优先级相同时会继续比较 task；字典不可排序，因此程序会抛出 TypeError。即便任务对象可以比较，它的自然顺序也不等于“先到先服务”。
解决办法是在中间加入单调序号：
(priority, sequence, task)优先级先决定顺序；优先级相同，较小的序号先出队；序号不会重复，所以比较永远不必走到任务对象。普通堆本身不保证稳定，这个序号才建立了稳定合同。
2. 支持更新与取消的完整实现
二叉堆擅长改根，却不擅长从数组中间找一个具名任务。下面用映射定位当前条目，取消时先标记，等条目来到根再跳过。这个策略叫惰性删除。数字越小越先出队；同分时按最近一次 add 的顺序排列。更新会获得新序号，所以同优先级更新相当于重新排队。
任务名必须是字符串，优先级必须是整数，检查发生在改动旧条目之前。不存在的任务名传给 remove，或没有活任务时调用 pop，均抛出 KeyError。此教学类不提供线程同步。
from __future__ import annotations
import heapq
import itertools
from operator import index as integer_index
from collections.abc import Iterable
from typing import Any, TypeVar
T = TypeVar("T")
class StablePriorityQueue:
def __init__(self) -> None:
self._heap: list[list[Any]] = []
self._entries: dict[str, list[Any]] = {}
self._counter = itertools.count()
self._removed = object()
def __len__(self) -> int:
return len(self._entries)
def add(self, name: str, payload: Any, priority: int = 0) -> None:
if not isinstance(name, str):
raise TypeError("name must be a string")
if not isinstance(priority, int):
raise TypeError("priority must be an integer")
if name in self._entries:
self.remove(name)
entry = [priority, next(self._counter), name, payload]
self._entries[name] = entry
heapq.heappush(self._heap, entry)
def remove(self, name: str) -> None:
if not isinstance(name, str):
raise TypeError("name must be a string")
entry = self._entries.pop(name)
entry[2] = self._removed
entry[3] = None
def compact(self) -> None:
self._heap = list(self._entries.values())
heapq.heapify(self._heap)
def pop(self) -> tuple[str, Any, int]:
while self._heap:
priority, _sequence, name, payload = heapq.heappop(self._heap)
if name is not self._removed:
del self._entries[name]
return name, payload, priority
raise KeyError("pop from an empty priority queue")
def top_k_largest(values: Iterable[T], k: int) -> list[T]:
k = integer_index(k)
if k <= 0:
return []
heap: list[T] = []
for value in values:
if len(heap) < k:
heapq.heappush(heap, value)
elif heap[0] < value:
heapq.heapreplace(heap, value)
return sorted(heap, reverse=True)
def merge_k_sorted(groups: list[list[T]]) -> list[T]:
heap: list[tuple[T, int, int]] = []
for group_index, group in enumerate(groups):
if group:
heapq.heappush(heap, (group[0], group_index, 0))
merged: list[T] = []
while heap:
value, group_index, item_index = heapq.heappop(heap)
merged.append(value)
next_index = item_index + 1
if next_index < len(groups[group_index]):
next_value = groups[group_index][next_index]
heapq.heappush(heap, (next_value, group_index, next_index))
return merged
if __name__ == "__main__":
queue = StablePriorityQueue()
queue.add("backup", {"node": 3}, priority=2)
queue.add("report", {"node": 1}, priority=2)
queue.add("alert", {"node": 7}, priority=1)
print(queue.pop())
print(queue.pop())
print(top_k_largest([3, 2, 1, 5, 6, 4], 2))
print(merge_k_sorted([[1, 4, 5], [1, 3, 4], [2, 6]]))完整输出如下，字典载荷从未互相比较：
('alert', {'node': 7}, 1)
('backup', {'node': 3}, 2)
[6, 5]
[1, 1, 2, 3, 4, 4, 5, 6]更新任务时，旧条目仍留在堆数组中，取消会清掉它对载荷的引用。len(queue) 只统计活任务。compact() 从活条目重建堆，保留原序号，所以不会改变同分顺序；何时调用取决于内存和延迟目标，本例没有自动阈值。
设堆数组含 m 条记录，其中 a 条活跃。按平均字典访问成本计，标记删除为 O(1)；加入的堆调整为 O(log(m+1))；一次出队若跳过 s 条陈旧记录，需要至多 O((s+1) log(m+1)) 的堆调整。底层列表和字典扩缩容还需按操作序列摊销。即使只剩一个活任务，一次 pop 也可能清理大量旧条目。重建活堆用 O(a)，释放旧列表还需处理 m 条引用；总成本为 O(m+a)，存储由 O(m+a) 回到 O(a)，均不计载荷对象。
3. Top K：保留候选集合的淘汰线
分数逐项到达，只保留最大的 k 个。用大小至多为 k 的最小堆，堆顶就是候选集合的淘汰线：
- 堆未满，直接加入；
- 新值不超过堆顶，它进不了前
k； - 新值超过堆顶，用它替换当前最小候选。
k 必须支持整数索引，浮点数会在读取输入前被拒绝。k <= 0 直接返回 []，不消耗输入；k >= n 返回全部值的降序副本；重复次数保留。对 k > 0，令 q=min(k,n)，扫描与最终排序的总上界是 O(n log(q+1))，辅助堆和结果各占 O(q)；k=1 仍需线性扫描。
元素应满足一致的顺序比较，不能含 NaN，比较成本按常数计。Top K 只承诺值和重复次数；相等对象之间的原始次序不保证保留。
这里用 heapreplace，是因为代码已经证明新值大于堆顶。它先移除堆中的根，再加入新值。heappushpop(heap, value) 则先考虑新值和旧根，返回两者中更小的值，把较大者留在堆里。固定大小筛选中两者都可能有用，但语义不能凭名字猜。
4. K 路归并：每队只派一个代表
若干输入列表已经按升序排好。先把每个非空列表的第一项放入堆；取走一项后，只补入同一列表的下一项。输入列表不会被修改，但也不会被预先检查是否有序：这是调用前提。
设共 K 个列表、N 个元素，其中 r 个列表非空。扫描列表头还要 O(K)，逐个加入初始堆为 O(r log(r+1))，后续处理为 O(N log(r+1))。由于 r <= N，总上界可写为 O(K + N log(r+1))；K=1 或大量空列表时也成立。辅助堆占 O(r)，返回列表另占 Θ(N)。
元素必须具有一致的比较与相等语义。值相等时，group_index 让编号较小的列表先出，同一列表保持原顺序。这也解释了元组中的平局规则。需要边读边输出时，可改成生成器，或使用 heapq.merge。
5. 三种标准库的方向不同
| 环境 | 默认根 / 头部 | 重要合同 |
|---|---|---|
Python heapq | 最小值 | 在普通 list 上维护堆；heapify 是线性时间 |
Java PriorityQueue | 比较器下的最小值 | 同值头部任意；迭代器不保证顺序；不允许 null；非同步 |
C++ std::priority_queue<T> | 默认最大值 | 容器适配器；用 std::greater<T> 得到最小优先队列 |
Python 3.14 新增 heapify_max、heappush_max、heappop_max 等公开最大堆 API。需要兼容更早版本时，对数值可取负数或包装比较键；Python 整数取负不会发生固定宽度溢出，但在其他语言中要检查最小整数的取负边界。heapq 只是作用于列表的算法模块；线程间阻塞式任务队列应看 queue.PriorityQueue。
Java 的 offer、poll 是 O(log n)，peek 是 O(1)。相同最小键的头部任意，不能把 PriorityQueue 当稳定队列。iterator() 和 toString() 展示的内部顺序也不是排序结果；需要有序取出时反复 poll，但这会清空队列。多线程修改要使用合适的同步或 PriorityBlockingQueue。
C++ 默认比较器 std::less<T> 让最大值出现在 top()。Top K 最大值问题需要维护最小堆：
#include <functional>
#include <queue>
#include <vector>
std::priority_queue<int, std::vector<int>, std::greater<int>> candidates;标准适配器提供 push、pop、top，不提供有序迭代或任意元素删除。需要更新优先级时，常见做法也是加入新条目并在弹出时跳过旧版本，或另选支持句柄的数据结构。
6. 延伸模式：两个堆维护中位数
数据流中位数把较小的一半放在最大堆，较大的一半放在最小堆，并保持两边大小差不超过 1。奇数个值时，中位数是较大那一堆的堆顶；偶数个数值时，通常取两个堆顶的平均值。空输入时尚无中位数。这是“只维护边界，不维护完整顺序”的另一个例子。先把单堆合同练熟，再实现这个扩展。
7. 故障排查
- 同优先级偶发
TypeError：元组缺少唯一序号，比较落到了载荷。 - 取消后仍出队：
pop没有循环跳过删除标记。 - 更新后活任务数增加：替换前没有移除同名旧条目。
- Java 打印结果看似乱序：内部数组遍历没有排序保证。
- C++ Top K 淘汰了最大值：误用了默认最大优先队列。
- 内存持续增长：惰性删除留下太多陈旧条目，需要重建策略。
官方合同
- Python 3.14
heapq - Python
queue.PriorityQueue - Oracle Java 21
PriorityQueue - C++ draft:
priority_queue - C++ draft: heap operations
从任务顺序到图遍历
优先队列的工程难点往往不在上滤和下滤，而在业务顺序：数字越小是否越紧急，同分是否先到先出，更新与取消是否允许，空队列怎样报告。先写清接口合同，再让堆执行它；若采用惰性删除，还要监控陈旧条目并安排重建。
下一章进入 图，路径不再只有父子方向，待处理节点也会从一条队列扩展成整张网络。