数据结构 / 06
堆：让“下一次取最值”变得便宜
如果任务需要不断加入新元素，又要反复取出当前最小或最大值，先排序再取值会重复做大量工作。堆只维护“堆顶是当前极值”这一条不变量，把插入和删除堆顶控制在 O(log n)，因此适合优先队列、Top K、定时任务和多路归并。
阅读位置
本篇承接 栈与队列，先复用前置结构的接口和不变量，再学习当前结构的新增能力。读完后继续到 图，把当前结构放进更大的关系或构造问题中。
本节知识地图
mindmap
root((堆))
结构
完全二叉树
数组存储
父子下标
不变量
最小堆
最大堆
只保证堆顶
操作
peek
push
pop
heapify
replace
应用
优先队列
Top K
多路归并
对顶堆
边界
空堆
重复优先级
更新和删除
接口契约
堆对外至少提供 peek、push、pop 和 heapify。空堆行为、重复优先级和是否原地修改必须写入契约；复杂度分别按最坏或均摊口径说明。
class MinHeap:
def __init__(self, values=()):
self.data = list(values)
self.heapify()
def heapify(self):
for i in range(len(self.data) // 2 - 1, -1, -1):
self._sift_down(i)
def push(self, value):
self.data.append(value)
self._sift_up(len(self.data) - 1)
def pop(self):
if not self.data:
raise IndexError("pop from empty heap")
self.data[0], last = self.data[-1], self.data[0]
self.data.pop()
if self.data:
self._sift_down(0)
return last
def _sift_up(self, i):
while i:
p = (i - 1) // 2
if self.data[p] <= self.data[i]:
break
self.data[p], self.data[i] = self.data[i], self.data[p]
i = p
def _sift_down(self, i):
n = len(self.data)
while 2 * i + 1 < n:
child = 2 * i + 1
if child + 1 < n and self.data[child + 1] < self.data[child]:
child += 1
if self.data[i] <= self.data[child]:
break
self.data[i], self.data[child] = self.data[child], self.data[i]
i = child
def heap_sort(values):
heap = MinHeap(values)
return [heap.pop() for _ in range(len(heap.data))]
一、堆维护什么性质
堆是完全二叉树：除最后一层外每层都填满，最后一层从左到右连续填充。完全性让它可以紧凑地放进数组，不需要为每个节点保存左右指针。
最小堆要求每个父节点不大于子节点，所以根是全局最小值；最大堆反之。堆不是有序数组：除根和局部父子关系外，兄弟节点及不同子树之间没有排序保证。
对 0-based 数组，节点 i 的关系是：
parent(i) = (i - 1) // 2
left(i) = 2 * i + 1
right(i) = 2 * i + 2
只要每次公开操作结束后恢复父子堆序，data[0] 就能可靠地提供极值。
二、插入：先放末尾，再上浮
新元素先放到数组末尾，这样不会破坏完全二叉树结构。它可能小于父节点，于是不断与父节点交换，直到到达根或父节点已经不大于它。沿树高移动，复杂度为 O(log n)。
def push(heap, value):
heap.append(value)
i = len(heap) - 1
while i:
p = (i - 1) // 2
if heap[p] <= heap[i]:
break
heap[p], heap[i] = heap[i], heap[p]
i = p
三、删除堆顶：末尾元素下沉
删除根后，若直接留下空洞就破坏完全性。把末尾元素移到根，再与较小的子节点交换，直到两个子节点都不小于它。每次只沿一条树高路径移动，复杂度是 O(log n)。
def pop_min(heap):
if not heap:
raise IndexError('pop from empty heap')
result = heap[0]
last = heap.pop()
if heap:
heap[0] = last
i = 0
while True:
child = 2 * i + 1
if child >= len(heap):
break
if child + 1 < len(heap) and heap[child + 1] < heap[child]:
child += 1
if heap[i] <= heap[child]:
break
heap[i], heap[child] = heap[child], heap[i]
i = child
return result
下沉时必须选择更小的子节点。若先和较大的子节点交换，较小的子节点仍可能违反堆序。
四、建堆为什么是 O(n)
逐个插入需要 O(n log n)。自底向上建堆从最后一个非叶节点开始下沉。靠近叶子的节点数量多但最多移动一两层，靠近根的节点少但可能移动较远；把所有移动距离相加，总复杂度是 O(n)。
def heapify(a):
for i in range(len(a) // 2 - 1, -1, -1):
sift_down(a, i)
heapify 通常原地修改输入并返回 None。调用者必须明确这一副作用，不能写成 heap = heapify(values) 后误以为得到新列表。
五、优先队列：堆的直接应用
优先队列只承诺下一次取出优先级最高或最低的元素。插入任务调用 push，处理任务调用 pop，查看下一任务调用 peek。相同优先级是否 FIFO（First In, First Out，先进先出）不是堆自动保证的。
Python 的 heapq 是最小堆。需要最大堆时，可以把数值取反，或把优先级和递增计数器放进元组：
import heapq
queue = []
counter = 0
heapq.heappush(queue, (priority, counter, task))
counter += 1
priority, _, task = heapq.heappop(queue)
元组前字段相同后会继续比较后字段，所以 task 不可比较时必须放入唯一 counter，避免运行时比较异常。
六、Top K：保留较小的候选堆
求最大的 K 个元素时，维护一个大小不超过 K 的最小堆。新元素比堆顶大，才替换堆顶；最终堆中留下最大的 K 个元素。每个元素最多进行一次 O(log K) 调整，总复杂度 O(n log K)，空间 O(K)。
求最小的 K 个元素则使用最大堆，思路完全对称。若 K 接近 n，完整排序的常数可能更好；选算法时要结合 K、数据是否流式到达以及是否需要完整有序输出。
七、多路归并：堆保存每路当前最小值
合并 K 个有序序列时，不需要把所有元素一次性放进数组。先把每个序列的第一个元素放入堆，并记录它来自哪一路；每次取出全局最小值，再把同一路的下一个元素加入堆。堆大小最多 K，总复杂度为 O(n log K)。
这个模式的关键不是“堆能排序”，而是堆只保存每条有序输入的一个候选。每条输入的局部顺序，帮助我们安全地丢弃已经取出的元素。
八、对顶堆维护动态中位数
把较小的一半放入最大堆，把较大的一半放入最小堆，并保持两个堆大小相差不超过一。这样中位数始终位于一个或两个堆顶。插入一个数后先按值放入对应堆，再平衡大小，复杂度 O(log n)。
对顶堆维护的是“分界线”而非完整排序。它适合在线查询中位数，但如果需要任意排名、区间中位数或删除任意值，堆可能不是最合适的结构。
九、更新和删除任意元素
堆能在 O(log n) 内更新已知索引的元素：值变小就上浮，变大就下沉。若只知道 value，先搜索它需要 O(n)；因此“堆支持任意删除 O(log n)”缺少已知位置这一前提。
需要按任务 ID 更新时，可额外维护 task_id → heap index 的映射。每次交换堆元素时同步更新映射，定位变成 O(1)，调整仍是 O(log n)，代价是实现和一致性维护更复杂。
常见误区
堆顶是极值不代表数组有序；不能用堆数组直接当排序结果。建堆的 O(n) 也只适用于自底向上 heapify，逐个 push 仍是 O(n log n)。
优先级相同的元素默认没有稳定顺序；需要 FIFO 时要显式加入 counter。空堆的 peek 和 pop 行为也必须在接口中约定，不能依赖异常“碰巧出现”。
面试表达
先说堆是完全二叉树，通常用数组存储，只维护父子堆序，因此堆顶能在 O(1) 取得极值。插入从末尾上浮，删除堆顶把末尾移到根再下沉，均为 O(log n)；自底向上建堆是 O(n)。随后根据场景说明固定大小堆求 Top K、多路归并和对顶堆维护中位数。
理解检查
- 为什么堆可以用数组而不需要左右指针？
- 删除堆顶时为什么要选择较小的子节点交换？
- 自底向上建堆为什么不是
O(n log n)？ - Top K 什么时候使用大小为 K 的堆？
- 为什么优先队列需要额外 counter 才能稳定 FIFO？
可观察实验
实现最小堆并在每次 push、pop 后检查所有父节点是否不大于子节点。分别用逐个插入和 heapify 构造不同规模数组，记录比较次数；再用固定大小堆处理流式 Top K，观察空间是否保持在 O(K)。
术语卡片
| 术语 | 英文或含义 | 本章定义 |
|---|---|---|
| Heap | 堆 | 维护父子堆序的完全二叉树 |
| Min-Heap | 最小堆 | 堆顶是最小值 |
| Max-Heap | 最大堆 | 堆顶是最大值 |
| FIFO | First In, First Out | 先进先出 |
| Top K | 前 K 个 | 只保留指定排名范围的元素 |
实现附录：完整代码与边界验证
前面的章节先建立模型；本附录集中保留实现级细节，便于需要手写代码时逐项核对。附录中的每个小节都服务于同一个目标：验证堆序不变量、接口边界和复杂度假设。
Python heapq 模块
Python 标准库 heapq 提供了最小堆的实现，直接操作普通列表。
heapq 的异常与副作用
| 调用 | 语义 |
|---|---|
heappop([]) |
抛 IndexError |
heap[0]（空列表） |
抛 IndexError |
heapify(a) |
原地修改 a，返回 None |
heappush(heap, x) |
原地修改 heap，返回 None |
nlargest(k, iterable) |
返回新列表，不修改原 iterable |
heapq 不封装 size、锁、重复值策略和业务错误；这些是调用者或更高层容器的责任。
用数组手写最小堆
1. 数组下标关系
对 0-based 数组：
parent(i) = (i - 1) // 2
left(i) = 2 * i + 1
right(i) = 2 * i + 2
2. 堆序不变量
heap[parent(i)] <= heap[i]
只要每次公开操作结束后维持这个不变量，heap[0] 就是最小值。
3. 完整实现
class MinHeap:
def __init__(self, values=()):
self._data = list(values)
self._heapify()
def __len__(self):
return len(self._data)
def is_empty(self):
return not self._data
def peek(self):
if not self._data:
raise IndexError("peek from empty heap")
return self._data[0]
def push(self, value):
self._data.append(value)
self._sift_up(len(self._data) - 1)
def pop(self):
if not self._data:
raise IndexError("pop from empty heap")
self._data[0], last = self._data[-1], self._data[0]
self._data.pop()
if self._data:
self._sift_down(0)
return last
def replace(self, value):
if not self._data:
raise IndexError("replace on empty heap")
result = self._data[0]
self._data[0] = value
self._sift_down(0)
return result
def _heapify(self):
for index in range(len(self._data) // 2 - 1, -1, -1):
self._sift_down(index)
def _sift_up(self, index):
while index > 0:
parent = (index - 1) // 2
if self._data[parent] <= self._data[index]:
break
self._data[parent], self._data[index] = (
self._data[index], self._data[parent]
)
index = parent
def _sift_down(self, index):
size = len(self._data)
while True:
smallest = index
left = index * 2 + 1
right = left + 1
if left < size and self._data[left] < self._data[smallest]:
smallest = left
if right < size and self._data[right] < self._data[smallest]:
smallest = right
if smallest == index:
return
self._data[index], self._data[smallest] = (
self._data[smallest], self._data[index]
)
index = smallest
手写实现的接口不变量
_data的长度就是堆 size。- 根节点位于
_data[0]。 - 每个父节点都不大于两个子节点。
push先放末尾再上浮，pop先把末尾移到根再下沉。_heapify从最后一个非叶节点倒序处理。
heapify 为什么比逐个 push 快
逐个 push 是：
第 1 个元素 O(1)
第 2 个元素 O(log 2)
...
第 n 个元素 O(log n)
总计 O(n log n)
自底向上的 heapify 让大多数低层节点只下沉很短距离，总成本 O(n)。这是堆面试题中经常追问的复杂度差异。
删除任意位置
如果已经知道要删除的索引：
def remove_at(heap, index):
if not 0 <= index < len(heap):
raise IndexError("heap index out of range")
last = heap.pop()
if index == len(heap):
return
heap[index] = last
parent = (index - 1) // 2
if index > 0 and heap[index] < heap[parent]:
# 向上调整
while index > 0:
parent = (index - 1) // 2
if heap[parent] <= heap[index]:
break
heap[parent], heap[index] = heap[index], heap[parent]
index = parent
else:
# 向下调整
size = len(heap)
while True:
child = index * 2 + 1
if child >= size:
break
if child + 1 < size and heap[child + 1] < heap[child]:
child += 1
if heap[index] <= heap[child]:
break
heap[index], heap[child] = heap[child], heap[index]
index = child
若只有 value 没有 index，先搜索 value 需要 O(n)，所以“堆支持任意删除 O(log n)”是不完整的说法。
更新优先级
更新一个元素可能变大也可能变小：
- 变小：向上调整。
- 变大：向下调整。
若需要按任务 ID 更新，通常维护：
task_id -> heap index
每次交换元素时同步更新 index map，这就是 indexed heap；它把更新定位从 O(n) 降到 O(log n)，但实现复杂度更高。
自定义 key
Python heapq 比较元组，不接受类似 sort(key=...) 的直接 key 参数。通用包装方式：
class Prioritized:
def __init__(self, priority, value):
self.priority = priority
self.value = value
def __lt__(self, other):
return self.priority < other.priority
实际项目更常用 (priority, counter, value)，避免改写比较协议。
不稳定性
堆只保证优先级顺序，不保证同优先级元素的先后。若业务要求稳定 FIFO，必须显式加入 counter；不能依赖 Python 对象地址或字典顺序。
堆与排序的选择
| 需求 | 堆 | 排序 |
|---|---|---|
| 只要一个极值 | O(1) 取堆顶 | 先排序浪费 |
| 动态插入并持续取极值 | 合适 | 每次重排成本高 |
| 一次性求全部有序结果 | O(n log n) 反复 pop | 直接排序通常更简单 |
| Top K 且 K 很小 | O(n log k) | O(n log n) |
| 任意查找/删除 | 不擅长 | 也不擅长，考虑映射 |
4. 为什么建堆是 O(n)
不是对每个元素都调用 push。自底向上调整时：
- 叶子节点不需要下沉。
- 越靠近底部的节点高度越小。
- 高度为 h 的节点数量约为
n / 2^(h+1)。
总成本：
n/4 × 1 + n/8 × 2 + n/16 × 3 + ...
= O(n)
5. pop 为什么要把末尾移到堆顶
删除堆顶后，若直接留下空洞，数组的完全二叉树结构被破坏。把最后一个元素移到根，再向下交换，可以同时恢复：
- 数组无空洞。
- 完全二叉树形状。
- 父子堆序。
6. 不支持任意删除的原因
堆只对堆顶提供直接定位。删除任意值前，通常要先 O(n) 找到它，再 O(log n) 调整；因此如果业务频繁按 key 删除，应考虑索引堆、惰性删除或其他结构。
7. 惰性删除
优先队列常把 (priority, task_id, active) 放入堆：
- 更新任务时插入新版本。
- 旧版本标记失效。
- pop 时跳过失效项。
优点是更新简单；代价是堆中会积累垃圾，需要定期重建。
8. 稳定优先级队列
import heapq
from itertools import count
counter = count()
heap = []
def push_task(priority, task):
heapq.heappush(heap, (priority, next(counter), task))
def pop_task():
if not heap:
raise IndexError("pop from empty priority queue")
priority, _, task = heapq.heappop(heap)
return priority, task
堆排序
1. 思路
将数组建成堆，再不断取出堆顶：
def heap_sort(values):
heap = list(values)
heapq.heapify(heap)
return [heapq.heappop(heap) for _ in range(len(heap))]
这个版本使用 O(n) 额外空间。原地堆排序要在数组内建最大堆、交换根和末尾，再缩小有效区间：
def heap_sort_in_place(values):
def sift_down(start, end):
root = start
while 2 * root + 1 <= end:
child = 2 * root + 1
if child + 1 <= end and values[child] < values[child + 1]:
child += 1
if values[root] >= values[child]:
return
values[root], values[child] = values[child], values[root]
root = child
n = len(values)
for start in range((n - 2) // 2, -1, -1):
sift_down(start, n - 1)
for end in range(n - 1, 0, -1):
values[0], values[end] = values[end], values[0]
sift_down(0, end - 1)
return values
原地堆排序 O(n log n)、额外空间 O(1)，但通常不稳定，缓存和常数也可能不如 Timsort。
2. 什么时候不选堆排序
- Python 业务代码通常优先
sorted，可读性和稳定性更好。 - 需要稳定排序时不能默认使用堆排序。
- 只需 Top K 时维护大小 K 的堆更省成本。
Top K 的两种策略
维护大小 K 的堆
遍历 n 个元素，堆始终不超过 K：
时间 O(n log k)
空间 O(k)
适合 n 很大、不能保存全部排序结果。
全量排序
时间 O(n log n)
空间依实现
适合需要完整有序结果或 K 接近 n。
频率题的 tie-break
题目常要求：
频率降序，频率相同按字典序升序
堆中应显式编码优先级；不能只把 (freq, word) 放入最小堆后期待得到正确输出。
优先队列的生命周期
取消任务
heapq 没有高效任意删除。常见惰性删除：
REMOVED = object()
counter = count()
heap = []
entry_finder = {}
def add_task(task, priority):
if task in entry_finder:
entry_finder[task][2] = REMOVED
entry = [priority, next(counter), task]
entry_finder[task] = entry
heapq.heappush(heap, entry)
def pop_task():
while heap:
priority, _, task = heapq.heappop(heap)
if task is not REMOVED:
entry_finder.pop(task, None)
return task
raise KeyError("priority queue is empty")
这段代码的接口语义是“同一 task 只保留最新版本”，旧 entry 仍占堆空间，垃圾过多时要重建。
线程安全
heapq 的函数本身不提供并发保护。多个线程共享堆时必须：
- 用锁包围复合操作。
- 或使用
queue.PriorityQueue。
后者更安全，但每次操作包含同步成本，不应直接拿来和裸 heapq 比基准。
堆的验证函数
def is_min_heap(values):
for parent, value in enumerate(values):
left = 2 * parent + 1
right = left + 1
if left < len(values) and value > values[left]:
return False
if right < len(values) and value > values[right]:
return False
return True
验证堆序不等于验证排序；例如 [1, 5, 3, 8, 7, 4] 可能是合法最小堆，但不是升序数组。
多路归并的接口
1. 为什么堆只放 K 个头
K 条有序序列中，下一次最小值只可能来自每条序列当前未取出的头部：
list 1: 1, 7, 10
list 2: 2, 4, 12
list 3: 3, 8, 9
堆中只需保存 (当前值, 序列编号, 位置)：
def merge_sorted_arrays(arrays):
heap = []
for array_id, array in enumerate(arrays):
if array:
heapq.heappush(heap, (array[0], array_id, 0))
result = []
while heap:
value, array_id, index = heapq.heappop(heap)
result.append(value)
next_index = index + 1
if next_index < len(arrays[array_id]):
next_value = arrays[array_id][next_index]
heapq.heappush(heap, (next_value, array_id, next_index))
return result
若总元素数为 N、序列数为 K，时间 O(N log K)，空间 O(K)（不计结果）。
2. 元组 tie-break
array_id 不是算法数据，而是为了：
- 值相同时稳定比较。
- 避免比较不可排序的数组对象。
- 找回下一元素的来源。
类似原则适用于合并链表和优先级任务。
堆与其他极值结构
| 结构 | 取极值 | 插入 | 删除任意值 | 是否有序 |
|---|---|---|---|---|
| 无序数组 | O(n) | O(1) | O(n) | 否 |
| 有序数组 | O(1) | O(n) | O(n) | 是 |
| 堆 | O(1) | O(log n) | O(n) 查找后 O(log n) | 只有局部序 |
| 平衡 BST | O(log n) | O(log n) | O(log n) | 是 |
若同时需要按 key 查找、按优先级取出和任意删除，单个堆往往不够，需要 HashMap + Heap 的组合。
堆的内部操作推演
上浮 sift-up
插入新值总是先放到数组末尾：
[1, 4, 3, 8, 7, 2]
^ 新值
若它小于父节点，就交换并继续向根移动：
[1, 4, 2, 8, 7, 3]
[1, 3, 2, 8, 7, 4]
最多移动树高 h=O(log n) 层。
下沉 sift-down
删除根后把最后一个元素放到根，再与更小的孩子交换：
[9, 4, 3, 8, 7] -> [3, 4, 9, 8, 7]
-> [3, 4, 7, 8, 9]
每层选择更小孩子，否则可能仍违反堆序。
为什么不能只和左孩子比较
若右孩子比左孩子更小，和左孩子交换后右侧仍可能比新父节点小。下沉必须在存在的孩子中选择极值。
优先队列的更新协议
版本号惰性删除
class TaskQueue:
def __init__(self):
self.heap = []
self.version = {}
def push(self, task, priority):
version = self.version.get(task, 0) + 1
self.version[task] = version
heapq.heappush(self.heap, (priority, version, task))
def pop(self):
while self.heap:
priority, version, task = heapq.heappop(self.heap)
if self.version.get(task) == version:
del self.version[task]
return task
raise KeyError("empty task queue")
新版本覆盖旧版本的逻辑有效性，旧条目直到 pop 才物理清理。
更新复杂度
- 无索引：插入新条目 O(log n)，旧条目惰性失效。
- 有索引堆：定位 O(1)，调整 O(log n)，实现需要同步交换索引。
- 直接在数组中搜索：更新定位 O(n)。
堆的边界检查
空堆 pop/peek
k=0、k>n、n=0
优先级相等且 payload 不可比较
重复值和重复任务
负数、浮点数和自定义对象比较
惰性删除垃圾条目过多
30 秒背诵
堆是用数组存储的完全二叉树，只保证父子堆序。push 末尾后 sift-up，pop 把末尾移到根后 sift-down，均为 O(log n)；自底向上 heapify 为 O(n)。它擅长动态取极值和 Top K，不擅长任意查找删除；优先级相同要加 counter，更新/取消可用索引堆或惰性删除。
堆的接口矩阵
| 需求 | 基础堆 | 需要的增强 |
|---|---|---|
| 查看极值 | peek |
无 |
| 动态插入 | push |
上浮 |
| 删除极值 | pop |
下沉 |
| 任意 key 更新 | 不直接支持 | index map |
| 取消任务 | 不直接支持 | 惰性删除/版本号 |
| 同优先级 FIFO | 不保证 | counter |
| 多线程共享 | 不安全 | 锁或 PriorityQueue |
“堆是有序的”只能理解为父子局部有序；要输出全序结果，应反复 pop 或直接排序，并写出 O(n log n) 成本。
堆接口边界测试
heap = MinHeap([5, 1, 3, 1])
assert is_min_heap(heap._data)
assert heap.peek() == 1
assert heap.pop() == 1
heap.push(0)
assert heap.pop() == 0
assert heap.pop() == 1
assert heap.pop() == 3
assert heap.pop() == 5
try:
heap.pop()
except IndexError:
pass
else:
raise AssertionError("empty heap should fail")
测试应同时验证输出顺序和每次公开操作后的堆序，不要只比较最终排序结果。
counter 保证同优先级任务按进入顺序比较，且不会比较不可排序的 task 对象。
基本操作
import heapq
heap = []
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
print(heap) # [1, 5, 3]
print(heapq.heappop(heap)) # 1（弹出最小值）
print(heap[0]) # 3（查看堆顶）
nums = [4, 1, 7, 3, 8, 5]
heapq.heapify(nums) # 原地建堆 O(n)
print(heapq.nsmallest(3, nums)) # [1, 3, 4]
print(heapq.nlargest(3, nums)) # [8, 7, 5]
提示：
nsmallest/nlargest在 K 很小时效率高；K 接近 n 时直接排序更快。
实现最大堆
heapq 只支持最小堆。实现最大堆的经典技巧是对值取反：
import heapq
max_heap = []
for val in [3, 1, 4, 1, 5, 9]:
heapq.heappush(max_heap, -val)
print(-heapq.heappop(max_heap)) # 9（弹出最大值）
print(-max_heap[0]) # 5（查看当前最大值）
自定义排序
用元组 (priority, data) 实现优先级队列。heapq 会按元组的第一个元素排序：
import heapq
tasks = []
heapq.heappush(tasks, (2, "写代码"))
heapq.heappush(tasks, (1, "修 Bug"))
heapq.heappush(tasks, (3, "开会"))
while tasks:
priority, task = heapq.heappop(tasks)
print(f"优先级 {priority}: {task}") # 修Bug -> 写代码 -> 开会
注意：优先级相同时 Python 会比较下一个元素。若元素不可比较，加递增计数器：
(priority, counter, data)。
高频题型
Top K 问题
Top K 是堆最经典的应用场景。核心思路：维护一个大小为 K 的堆。
LC 347 前 K 个高频元素
from collections import Counter
import heapq
def topKFrequent(nums, k):
count = Counter(nums)
# 简洁写法
return heapq.nlargest(k, count.keys(), key=count.get)
# 手动维护堆（面试更常考）
def topKFrequent_heap(nums, k):
count = Counter(nums)
heap = []
for num, freq in count.items():
heapq.heappush(heap, (freq, num))
if len(heap) > k:
heapq.heappop(heap)
return [num for freq, num in heap]
LC 215 数组中的第 K 个最大元素
思路：维护一个大小为 K 的最小堆，遍历完成后堆顶就是第 K 大元素。
def findKthLargest(nums, k):
heap = []
for num in nums:
heapq.heappush(heap, num)
if len(heap) > k:
heapq.heappop(heap)
return heap[0]
合并 K 个有序链表
LC 23 合并 K 个升序链表
思路：将每个链表头放入最小堆，每次弹出最小值，再将其 next 入堆。时间 O(N log K)。
def mergeKLists(lists):
heap = [(node.val, i, node) for i, node in enumerate(lists) if node]
heapq.heapify(heap)
dummy = curr = ListNode(0)
while heap:
val, i, node = heapq.heappop(heap)
curr.next = node
curr = curr.next
if node.next:
heapq.heappush(heap, (node.next.val, i, node.next))
return dummy.next
元组中
i用于打破值相同时的比较，避免ListNode之间无法比较。
数据流中位数
LC 295 数据流的中位数
思路：用两个堆维护数据流的左右两半：
- 大顶堆
max_heap：存较小的一半（取反模拟） - 小顶堆
min_heap：存较大的一半
始终保持 len(max_heap) == len(min_heap) 或 len(max_heap) == len(min_heap) + 1。
class MedianFinder:
def __init__(self):
self.small = [] # 大顶堆（取反）
self.large = [] # 小顶堆
def addNum(self, num):
heapq.heappush(self.small, -num)
heapq.heappush(self.large, -heapq.heappop(self.small))
if len(self.large) > len(self.small):
heapq.heappush(self.small, -heapq.heappop(self.large))
def findMedian(self):
if not self.small and not self.large:
raise IndexError("median of empty stream")
if len(self.small) > len(self.large):
return -self.small[0]
return (-self.small[0] + self.large[0]) / 2
滑动窗口最大值
LC 239 滑动窗口最大值
思路：用最大堆存储 (-nums[i], i)，每次取堆顶时检查索引是否在窗口内，不在则弹出。时间复杂度 O(n log n)。
def maxSlidingWindow(nums, k):
heap = []
result = []
for i in range(len(nums)):
heapq.heappush(heap, (-nums[i], i))
if i >= k - 1:
while heap[0][1] <= i - k:
heapq.heappop(heap)
result.append(-heap[0][0])
return result
补充：此题最优解是单调队列 O(n)，但堆解法更通用，面试中两种都应掌握。
边界测试清单
空堆 peek/pop/replace
单元素 push/pop
重复优先级和不可比较 payload
heapify 空列表和已排序列表
第 K 大的 k=1、k=n、非法 k
MedianFinder 空数据和偶数/奇数个数据
滑动窗口 k=1、k=len(nums)、k=0
惰性删除后堆中垃圾项
任意删除根、叶子和中间节点
面试表达：堆的实现
Q1：堆为什么用数组
堆是完全二叉树，按层序存入数组后不需要显式指针。0-based 下父节点是
(i-1)//2，左右孩子是2i+1和2i+2，连续内存也有更好的缓存局部性。
Q2：建堆为什么是 O(n)
自底向上只对非叶子节点下沉，底层节点数量多但高度小，高层节点少但高度大，按高度求和是 O(n)。逐个插入则每次可能上浮 log n，总成本 O(n log n)。
Q3：堆为什么不支持快速查任意值
堆只保证父子局部顺序，非堆顶节点之间没有全局排序。查找任意值最坏需要扫描 O(n)；知道索引后删除/更新才可 O(log n) 调整。
Q4：优先级相同为什么要 counter
Python 比较元组时会继续比较后续字段，payload 可能不可比较；而且业务可能要求同优先级 FIFO。加入递增 counter 同时解决可比较性和稳定性。
经典题目
Medium
| 题号 | 题目 | 关键思路 |
|---|---|---|
| 215 | 数组中的第 K 个最大元素 | 大小为 K 的最小堆 |
| 347 | 前 K 个高频元素 | 频率统计 + 堆 |
| 692 | 前 K 个高频单词 | 自定义排序的堆 |
| 378 | 有序矩阵中第 K 小的元素 | 多路归并 + 堆 |
Hard
| 题号 | 题目 | 关键思路 |
|---|---|---|
| 23 | 合并 K 个升序链表 | 堆维护 K 个链表头 |
| 295 | 数据流的中位数 | 对顶堆（大顶堆 + 小顶堆） |
| 239 | 滑动窗口最大值 | 最大堆 / 单调队列 |
| 632 | 最小区间 | 堆 + 滑动窗口 |
学习建议
- 堆是面试高频数据结构：几乎所有涉及”第 K 大/小”、”前 K 个”、”合并多路有序序列”的题目都可以用堆解决。
- 熟练掌握
heapqAPI：heappush、heappop、heapify、nlargest、nsmallest是基础工具。 - Top K 是必考题型：LC 215 和 LC 347 建议多写几遍，做到闭眼能写。
- 理解取反技巧：Python 没有内置最大堆，取反是万能方案。
- 对顶堆是进阶技巧：LC 295 的大顶堆 + 小顶堆模式在很多变体题中都会用到。
堆接口自测与面试复盘
堆序检查
def validate_min_heap(heap):
for index, value in enumerate(heap):
left = 2 * index + 1
right = left + 1
if left < len(heap):
assert value <= heap[left]
if right < len(heap):
assert value <= heap[right]
操作边界
- 空堆
peek/pop是否抛异常。 heapify([])是否允许。- 优先级相同的 payload 是否可比较。
- Top K 中 k=0、k=n、k>n 如何处理。
- 惰性删除的垃圾条目何时清理。
最小口述
堆是完全二叉树的数组表示，只维护父子堆序。插入末尾后上浮，删除堆顶后末尾补根再下沉，都是 O(log n)；自底向上建堆是 O(n)。堆适合动态取极值、Top K 和多路归并，但任意查找/删除需要额外索引或退化为 O(n)。
小结
堆的本质是一棵用数组存储的完全二叉树，核心价值在于高效获取极值。在 Python 中，heapq 模块提供了简洁的最小堆操作，配合取反技巧和元组排序即可应对绝大多数面试场景。掌握 Top K、多路归并、对顶堆三种模式，堆相关的题目就不再是难点。
堆终局：接口与不变量
数组形状：完全二叉树
堆序：父节点 <= 或 >= 子节点
极值：只保证堆顶
push：末尾加入后上浮
pop：末尾补根后下沉
heapify：自底向上 O(n)
任意 key：需要索引或惰性删除
同优先级：需要 counter
空堆：必须定义异常或返回值
只要业务需要全局有序、任意查找和任意删除，单个堆就不是完整答案。
堆的手写验收：下标与方向
下标公式必须统一
0-based 数组中，节点 i 的父节点是 (i - 1) // 2，左孩子是 2*i + 1，右孩子是 2*i + 2。1-based 写法虽然能少一次加法，但不能和 0-based 代码混用；混用会表现为某些小样例通过、大样例堆序被破坏。
上浮与下沉的终止条件
- 上浮：只要当前节点优于父节点就交换；到根或父节点已经满足堆序时停止。
- 下沉：先选两个孩子中更优的那个，再与当前节点比较；没有更优孩子时停止。
- 删除任意位置：用最后一个元素覆盖空洞，然后根据它与父节点的关系选择上浮或下沉，不能固定只做一种。
heapify 的 O(n) 解释
从最后一个非叶节点向前下沉。底层节点数量最多但高度很小，顶部节点数量少但高度较大，所有节点的“高度总和”是线性的，因此建堆是 O(n)，不是把 n 次 push 简单相加成 O(n log n)。
优先队列的稳定性
业务 payload 可能不可比较，且同优先级时常要求先进先出。把条目编码为 (priority, sequence, payload)：先按优先级，再按递增序号比较，payload 永远不会被拿来比较。序号溢出或多线程生成时，要明确是否使用单调原子计数器。
堆结构测试矩阵
[]、[x]
重复优先级和不可比较 payload
连续 push 后逐个 pop 是否有序
heapify 与逐个 push 的结果是否一致
remove_at(0)、remove_at(last)、越界
update 后向上和向下两种方向
Top K 的 k=0、k=1、k>=n
多路归并中的空迭代器
验收时不要只检查堆顶；应遍历所有父子下标验证堆序，并把 pop 序列与 sorted 结果对比。
复杂度口述的边界
peek 只读根节点，所以是 O(1)；push、pop 和已知下标的 update 是 O(log n)；remove(value) 若没有索引表，必须线性查找，成本 O(n)。heapify 是 O(n)，但复制输入数组的空间是否计入，要根据接口是原地还是返回副本来说明。
为什么堆不是排序数组
堆只保证父子局部有序，兄弟子树之间没有顺序。因此它能快速取最小值，却不能 O(1) 判断任意值是否存在，也不能按下标得到全局第 k 小。需要完整有序遍历时，应使用排序数组、平衡树或直接排序；需要动态极值时才选堆。
面试复述可以落到一句话：堆优化的是“取极值”，不是所有查询；任何额外能力都要靠索引表、计数器或惰性删除补上。
写比较器时先统一“更优”的方向：最小堆用小值上浮，最大堆用大值上浮；不要在 push 和 pop 中各写一套相反条件。
这条方向约定也适用于 Top K 和双堆中位数，先定义堆顶代表什么，再决定比较符号。
堆的验收器可以只依赖下标公式，不依赖 push/pop 实现本身；这样才能避免“用同一个错误验证另一个错误”。
for i in range((n - 2) // 2 + 1):
check(parent(i), left(i))
check(parent(i), right(i))
叶子节点无需下沉，最后一个非叶节点是 (n - 2) // 2；空堆和单元素堆要避免负下标误判。
验证 pop 序列时，允许相同优先级的 payload 顺序按计数器确定，但不能拿不可比较对象直接做二次比较。
从数组手算建堆
给定 [9, 4, 7, 1, 0, 3]
最后非叶节点 i=(n-2)//2=2
先下沉 7，再下沉 4，再下沉 9
每次只和更优孩子交换
最终逐个 pop 得到非降序序列
建堆过程中不能从根向下逐个插入，那是 O(n log n)；自底向上的顺序才利用了叶子已经天然成堆这一事实。
父子关系决定堆序
数组位置决定完全树形
删除后必须恢复两者
| ← 返回数据结构 | 上一篇：树与二叉树 | 下一篇：图 → |