堆 · Heap / Priority Queue

一句话骨架：堆是「能在 O(log n) 内拿到当前最值」的结构——Top-K 用「大小为 K 的最小堆」，中位数用「双堆对顶」，合并 K 路用「k 元素的最小堆」。

Python 速查

heapq 默认是最小堆：

import heapq
h = []
heapq.heappush(h, x)        # 推入
small = heapq.heappop(h)    # 弹出最小
heapq.heappushpop(h, x)     # 先 push 再 pop，O(log n) 一次完成
heapq.nlargest(k, nums)     # 直接拿前 K 大
heapq.heapify(arr)          # 把任意 list 原地变成堆，O(n)

要做最大堆？把值取负后再入堆：heappush(h, -x) / -heappop(h)。

关卡 1 · 数组中的第 K 个最大元素

学习目标

真正理解「找第 K 大用大小为 K 的最小堆」这一反直觉模板——堆顶就是答案；同时与排序法 / quickselect 三种解法做时间复杂度对比。

#215数组中的第 K 个最大元素中等

♥♥♥♥♥+0 XP🔥 0

0 / 12

📝 题目

给定整数数组 nums 和整数 k，请返回数组中第 k 个最大的元素。

请注意，你需要找的是数组排序后的第 k 个最大的元素，而不是第 k 个不同的元素。

你必须设计并实现时间复杂度为 O(n) 的算法解决此问题。

示例 1

输入nums = [3,2,1,5,6,4], k = 2

输出5

示例 2

输入nums = [3,2,3,1,2,4,5,5,6], k = 4

输出4

💡 第 4 大允许重复值各占一名

示例 3

输入nums = [1], k = 1

输出1

约束

• 1 ≤ k ≤ nums.length ≤ 10⁵
• -10⁴ ≤ nums[i] ≤ 10⁴

💡 思路：维护「大小为 k 的最小堆」：堆里始终保存「迄今为止最大的 k 个元素」，堆顶就是它们中的最小——也就是「第 k 大」。新元素若大于堆顶，就 heappushpop 把堆顶挤掉。时间 O(n log k)，比排序 O(n log n) 优。要严格 O(n) 需用 quickselect。

✅ 完整解法

时间 O(n log k) · 空间 O(k)

先把整段解法看懂，下面的选择题是对这段代码逐行的理解检验。

 1import heapq 2  3def findKthLargest(nums: list[int], k: int) -> int: 4    # 维护一个大小为 k 的最小堆——堆顶就是「目前为止第 k 大」 5    heap = [] 6    for x in nums: 7        if len(heap) < k: 8            heapq.heappush(heap, x) 9        elif x > heap[0]:10            heapq.heappushpop(heap, x)11    return heap[0]

🐞 单步可视化

示例：nums = [3, 2, 1, 5, 6, 4], k = 2

代码L5

 1import heapq 2  3def findKthLargest(nums: list[int], k: int) -> int: 4    # 维护一个大小为 k 的最小堆——堆顶就是「目前为止第 k 大」 5    heap = [] 6    for x in nums: 7        if len(heap) < k: 8            heapq.heappush(heap, x) 9        elif x > heap[0]:10            heapq.heappushpop(heap, x)11    return heap[0]

数据流

k2heap.size0

nums

0

3

1

2

2

1

3

5

4

6

5

4

最小堆（顶=最小）（顶 → 底）

空

1 / 14初始化空堆 heap = []

速度 慢中快

🧠 理解检验

每题都对应解法的某一行或某个决策

要找「第 k 大」，应该用大根堆还是小根堆？

Python heapq 默认是？

为什么用 heappushpop(heap, x) 而不是先 push 再 pop？

 1import heapq 2  3def findKthLargest(nums: list[int], k: int) -> int: 4    # 维护一个大小为 k 的最小堆——堆顶就是「目前为止第 k 大」 5    heap = [] 6    for x in nums: 7        if len(heap) < k: 8            heapq.heappush(heap, x) 9        elif x > heap[0]:10            heapq.heappushpop(heap, x)11    return heap[0]

为什么进入 elif 分支前要先判断 x > heap[0]？

 1import heapq 2  3def findKthLargest(nums: list[int], k: int) -> int: 4    # 维护一个大小为 k 的最小堆——堆顶就是「目前为止第 k 大」 5    heap = [] 6    for x in nums: 7        if len(heap) < k: 8            heapq.heappush(heap, x) 9        elif x > heap[0]:10            heapq.heappushpop(heap, x)11    return heap[0]

初始填充阶段（heap 还没满）应该怎么处理？

 1import heapq 2  3def findKthLargest(nums: list[int], k: int) -> int: 4    # 维护一个大小为 k 的最小堆——堆顶就是「目前为止第 k 大」 5    heap = [] 6    for x in nums: 7        if len(heap) < k: 8            heapq.heappush(heap, x) 9        elif x > heap[0]:10            heapq.heappushpop(heap, x)11    return heap[0]

循环结束后，第 k 大是堆里的哪个元素？

 1import heapq 2  3def findKthLargest(nums: list[int], k: int) -> int: 4    # 维护一个大小为 k 的最小堆——堆顶就是「目前为止第 k 大」 5    heap = [] 6    for x in nums: 7        if len(heap) < k: 8            heapq.heappush(heap, x) 9        elif x > heap[0]:10            heapq.heappushpop(heap, x)11    return heap[0]

本解法的时间复杂度是？

空间复杂度是？

题目要求 O(n)。能用堆达到 O(n) 吗？该用什么算法？

若直接 sorted(nums, reverse=True)[k-1]，复杂度是？

heapq.nlargest(k, nums) 与本写法的关系？

若题目改成「找第 k 小」，最少要改几处？

关卡 2 · 前 K 个高频元素

学习目标

学会把「频次」作为堆的比较 key——元组 (freq, num) 入堆是 Python 的标准技巧；并对照桶排序解法理解 O(n log k) 与 O(n) 的取舍。

#347前 K 个高频元素中等

♥♥♥♥♥+0 XP🔥 0

0 / 12

📝 题目

给你一个整数数组 nums 和一个整数 k，请你返回其中出现频率前 k 高的元素。

你可以按任意顺序返回答案。

示例 1

输入nums = [1,1,1,2,2,3], k = 2

输出[1,2]

💡 1 出现 3 次，2 出现 2 次

示例 2

输入nums = [1], k = 1

输出[1]

示例 3

输入nums = [4,1,-1,2,-1,2,3], k = 2

输出[-1, 2]

💡 负数也合法

约束

• 1 ≤ nums.length ≤ 10⁵
• -10⁴ ≤ nums[i] ≤ 10⁴
• k 在合法范围内（结果是唯一的）

💡 思路：两步：① `Counter` 统计每个值的频次；② 用大小为 k 的最小堆按频次维护「迄今最热门的 k 个」——堆顶是它们里频次最低的。新元素若频次大于堆顶就 heappushpop。返回时把堆里的 num 提取出来即可（顺序不重要）。时间 O(n log k)。

✅ 完整解法

时间 O(n log k) · 空间 O(n)

先把整段解法看懂，下面的选择题是对这段代码逐行的理解检验。

 1import heapq 2from collections import Counter 3  4def topKFrequent(nums: list[int], k: int) -> list[int]: 5    count = Counter(nums)  # {值: 出现次数} 6    # 用「大小为 k 的最小堆」按频次维护——堆顶=当前最不热门的 7    heap = []  # 元素形如 (freq, num) 8    for num, freq in count.items(): 9        if len(heap) < k:10            heapq.heappush(heap, (freq, num))11        elif freq > heap[0][0]:12            heapq.heappushpop(heap, (freq, num))13    return [num for freq, num in heap]

🐞 单步可视化

示例：nums = [1, 1, 1, 2, 2, 3], k = 2

代码L4

 1import heapq 2from collections import Counter 3  4def topKFrequent(nums: list[int], k: int) -> list[int]: 5    count = Counter(nums)  # {值: 出现次数} 6    # 用「大小为 k 的最小堆」按频次维护——堆顶=当前最不热门的 7    heap = []  # 元素形如 (freq, num) 8    for num, freq in count.items(): 9        if len(heap) < k:10            heapq.heappush(heap, (freq, num))11        elif freq > heap[0][0]:12            heapq.heappushpop(heap, (freq, num))13    return [num for freq, num in heap]

数据流

k2heap.size0count{}

nums

0

1

1

1

2

1

3

2

4

2

5

3

最小堆（顶=最不热门，元素=(freq,num)）（顶 → 底）

空

1 / 30函数入口：topKFrequent(nums, k)

速度 慢中快

🧠 理解检验

每题都对应解法的某一行或某个决策

第一步统计频次，应该用什么数据结构？

 1import heapq 2from collections import Counter 3  4def topKFrequent(nums: list[int], k: int) -> list[int]: 5    count = Counter(nums)  # {值: 出现次数} 6    # 用「大小为 k 的最小堆」按频次维护——堆顶=当前最不热门的 7    heap = []  # 元素形如 (freq, num) 8    for num, freq in count.items(): 9        if len(heap) < k:10            heapq.heappush(heap, (freq, num))11        elif freq > heap[0][0]:12            heapq.heappushpop(heap, (freq, num))13    return [num for freq, num in heap]

Counter(nums).most_common(k) 一行就能得到答案吗？

维护 top-k 高频，应该用大根堆还是小根堆？

堆里每个元素应当存什么？

 1import heapq 2from collections import Counter 3  4def topKFrequent(nums: list[int], k: int) -> list[int]: 5    count = Counter(nums)  # {值: 出现次数} 6    # 用「大小为 k 的最小堆」按频次维护——堆顶=当前最不热门的 7    heap = []  # 元素形如 (freq, num) 8    for num, freq in count.items(): 9        if len(heap) < k:10            heapq.heappush(heap, (freq, num))11        elif freq > heap[0][0]:12            heapq.heappushpop(heap, (freq, num))13    return [num for freq, num in heap]

若两个元素频次相同，元组 (freq, num) 怎么比？

heap[0][0] 取的是？

 1import heapq 2from collections import Counter 3  4def topKFrequent(nums: list[int], k: int) -> list[int]: 5    count = Counter(nums)  # {值: 出现次数} 6    # 用「大小为 k 的最小堆」按频次维护——堆顶=当前最不热门的 7    heap = []  # 元素形如 (freq, num) 8    for num, freq in count.items(): 9        if len(heap) < k:10            heapq.heappush(heap, (freq, num))11        elif freq > heap[0][0]:12            heapq.heappushpop(heap, (freq, num))13    return [num for freq, num in heap]

最后返回结果时为什么用列表推导 [num for freq, num in heap]？

 1import heapq 2from collections import Counter 3  4def topKFrequent(nums: list[int], k: int) -> list[int]: 5    count = Counter(nums)  # {值: 出现次数} 6    # 用「大小为 k 的最小堆」按频次维护——堆顶=当前最不热门的 7    heap = []  # 元素形如 (freq, num) 8    for num, freq in count.items(): 9        if len(heap) < k:10            heapq.heappush(heap, (freq, num))11        elif freq > heap[0][0]:12            heapq.heappushpop(heap, (freq, num))13    return [num for freq, num in heap]

本解法时间复杂度是？

空间复杂度是？

能否用「桶排序」做到 O(n)？

heappush(heap, (freq, num)) 与 heap.append((freq, num)); heapq.heapify(heap) 的差别？

若 k = 数组的 unique 值数量，本算法会怎样？

关卡 3 · 数据流的中位数

学习目标

掌握「双堆对顶」的不变量与 addNum 的中转写法——理解为什么不能直接 push 进对应一侧，以及 Python 没有原生大根堆时如何用「值取负」绕开。

#295数据流的中位数困难

♥♥♥♥♥+0 XP🔥 0

0 / 12

📝 题目

中位数是有序整数列表中最中间的数。如果列表的长度是偶数，中位数则是中间两个数的平均值。

请设计一个支持以下两种操作的数据结构：
- addNum(num)：将整数 num 加入数据结构
- findMedian()：返回当前所有元素的中位数

要求 addNum 与 findMedian 的复杂度都尽可能低。

示例 1

输入addNum(1); addNum(2); findMedian(); addNum(3); findMedian();

输出1.5, 2.0

💡 加入 1, 2 → 中位数 1.5；再加入 3 → [1,2,3] → 中位数 2

示例 2

输入addNum(-1); addNum(-2); findMedian(); addNum(-3); findMedian(); addNum(-4); findMedian();

输出-1.5, -2.0, -2.5

约束

• -10⁵ ≤ num ≤ 10⁵
• 至少在 findMedian 之前会调用一次 addNum
• 至多调用 5·10⁴ 次

💡 思路：「双堆对顶」：大根堆 low 装较小的一半，小根堆 high 装较大的一半。维持不变量 `len(low) - len(high) ∈ {0, 1}`。中位数永远在堆顶——奇数时取 low 顶，偶数时取两堆顶平均。Python 没有原生大根堆，把值取负塞进 heapq 即可。每次 addNum 是 O(log n)，findMedian 是 O(1)。

✅ 完整解法

时间 addNum O(log n), findMedian O(1) · 空间 O(n)

先把整段解法看懂，下面的选择题是对这段代码逐行的理解检验。

 1import heapq 2  3class MedianFinder: 4    def __init__(self): 5        # low: 大根堆（存较小的一半），用「负数」实现大根堆 6        self.low = [] 7        # high: 小根堆（存较大的一半） 8        self.high = [] 9 10    def addNum(self, num: int) -> None:11        # 默认让 low 比 high 多一个；偶数时两者等长12        if len(self.low) == len(self.high):13            # 先入 high 中转，再把 high 顶给 low——保证 low 的最大 ≤ high 的最小14            heapq.heappush(self.high, num)15            heapq.heappush(self.low, -heapq.heappop(self.high))16        else:17            heapq.heappush(self.low, -num)18            heapq.heappush(self.high, -heapq.heappop(self.low))19 20    def findMedian(self) -> float:21        if len(self.low) > len(self.high):22            return -self.low[0]  # 注意取负还原23        return (-self.low[0] + self.high[0]) / 2

🐞 单步可视化

示例：流入 [1, 2, 3, 4, 5, 6, 7]

代码L4

 1import heapq 2  3class MedianFinder: 4    def __init__(self): 5        # low: 大根堆（存较小的一半），用「负数」实现大根堆 6        self.low = [] 7        # high: 小根堆（存较大的一半） 8        self.high = [] 9 10    def addNum(self, num: int) -> None:11        # 默认让 low 比 high 多一个；偶数时两者等长12        if len(self.low) == len(self.high):13            # 先入 high 中转，再把 high 顶给 low——保证 low 的最大 ≤ high 的最小14            heapq.heappush(self.high, num)15            heapq.heappush(self.low, -heapq.heappop(self.high))16        else:17            heapq.heappush(self.low, -num)18            heapq.heappush(self.high, -heapq.heappop(self.low))19 20    def findMedian(self) -> float:21        if len(self.low) > len(self.high):22            return -self.low[0]  # 注意取负还原23        return (-self.low[0] + self.high[0]) / 2

数据流

low.size0high.size0

数据流（按顺序 addNum）

0

1

1

2

2

3

3

4

4

5

5

6

6

7

双堆 [high 小根堆 | low 大根堆]（顶 → 底）

── 中位线 ──

1 / 37初始化：low 大根堆（存负数实现）、high 小根堆，全空

速度 慢中快

🧠 理解检验

每题都对应解法的某一行或某个决策

为什么要用「双堆」而不是「一个有序列表 + 二分插入」？

low 和 high 应该分别是？

 1import heapq 2  3class MedianFinder: 4    def __init__(self): 5        # low: 大根堆（存较小的一半），用「负数」实现大根堆 6        self.low = [] 7        # high: 小根堆（存较大的一半） 8        self.high = [] 9 10    def addNum(self, num: int) -> None:11        # 默认让 low 比 high 多一个；偶数时两者等长12        if len(self.low) == len(self.high):13            # 先入 high 中转，再把 high 顶给 low——保证 low 的最大 ≤ high 的最小14            heapq.heappush(self.high, num)15            heapq.heappush(self.low, -heapq.heappop(self.high))16        else:17            heapq.heappush(self.low, -num)18            heapq.heappush(self.high, -heapq.heappop(self.low))19 20    def findMedian(self) -> float:21        if len(self.low) > len(self.high):22            return -self.low[0]  # 注意取负还原23        return (-self.low[0] + self.high[0]) / 2

Python 的 heapq 是最小堆。要做大根堆怎么办？

 1import heapq 2  3class MedianFinder: 4    def __init__(self): 5        # low: 大根堆（存较小的一半），用「负数」实现大根堆 6        self.low = [] 7        # high: 小根堆（存较大的一半） 8        self.high = [] 9 10    def addNum(self, num: int) -> None:11        # 默认让 low 比 high 多一个；偶数时两者等长12        if len(self.low) == len(self.high):13            # 先入 high 中转，再把 high 顶给 low——保证 low 的最大 ≤ high 的最小14            heapq.heappush(self.high, num)15            heapq.heappush(self.low, -heapq.heappop(self.high))16        else:17            heapq.heappush(self.low, -num)18            heapq.heappush(self.high, -heapq.heappop(self.low))19 20    def findMedian(self) -> float:21        if len(self.low) > len(self.high):22            return -self.low[0]  # 注意取负还原23        return (-self.low[0] + self.high[0]) / 2

本实现的不变量是？

当 len(low) == len(high) 时来一个新数，为什么不能直接 push 进 low？

 1import heapq 2  3class MedianFinder: 4    def __init__(self): 5        # low: 大根堆（存较小的一半），用「负数」实现大根堆 6        self.low = [] 7        # high: 小根堆（存较大的一半） 8        self.high = [] 9 10    def addNum(self, num: int) -> None:11        # 默认让 low 比 high 多一个；偶数时两者等长12        if len(self.low) == len(self.high):13            # 先入 high 中转，再把 high 顶给 low——保证 low 的最大 ≤ high 的最小14            heapq.heappush(self.high, num)15            heapq.heappush(self.low, -heapq.heappop(self.high))16        else:17            heapq.heappush(self.low, -num)18            heapq.heappush(self.high, -heapq.heappop(self.low))19 20    def findMedian(self) -> float:21        if len(self.low) > len(self.high):22            return -self.low[0]  # 注意取负还原23        return (-self.low[0] + self.high[0]) / 2

addNum 的「中转」做法（先入对面堆再把对面堆顶转过来）目的是？

findMedian 在偶数总数下的返回值是？

 1import heapq 2  3class MedianFinder: 4    def __init__(self): 5        # low: 大根堆（存较小的一半），用「负数」实现大根堆 6        self.low = [] 7        # high: 小根堆（存较大的一半） 8        self.high = [] 9 10    def addNum(self, num: int) -> None:11        # 默认让 low 比 high 多一个；偶数时两者等长12        if len(self.low) == len(self.high):13            # 先入 high 中转，再把 high 顶给 low——保证 low 的最大 ≤ high 的最小14            heapq.heappush(self.high, num)15            heapq.heappush(self.low, -heapq.heappop(self.high))16        else:17            heapq.heappush(self.low, -num)18            heapq.heappush(self.high, -heapq.heappop(self.low))19 20    def findMedian(self) -> float:21        if len(self.low) > len(self.high):22            return -self.low[0]  # 注意取负还原23        return (-self.low[0] + self.high[0]) / 2

findMedian 在奇数总数下应该返回？

 1import heapq 2  3class MedianFinder: 4    def __init__(self): 5        # low: 大根堆（存较小的一半），用「负数」实现大根堆 6        self.low = [] 7        # high: 小根堆（存较大的一半） 8        self.high = [] 9 10    def addNum(self, num: int) -> None:11        # 默认让 low 比 high 多一个；偶数时两者等长12        if len(self.low) == len(self.high):13            # 先入 high 中转，再把 high 顶给 low——保证 low 的最大 ≤ high 的最小14            heapq.heappush(self.high, num)15            heapq.heappush(self.low, -heapq.heappop(self.high))16        else:17            heapq.heappush(self.low, -num)18            heapq.heappush(self.high, -heapq.heappop(self.low))19 20    def findMedian(self) -> float:21        if len(self.low) > len(self.high):22            return -self.low[0]  # 注意取负还原23        return (-self.low[0] + self.high[0]) / 2

addNum 的时间复杂度是？

空间复杂度是？

若题目改为 99% 的输入在 [0, 100]，能怎么优化？

若改成「滑动窗口的中位数」（最近 k 个数），双堆还够用吗？

同 pattern 索引（hot100 同类题）

题号	题名	难度	关键变形
23	合并 K 个升序链表	困难	k 元素的最小堆，每次弹出最小并把它的 next 入堆
239	滑动窗口最大值	困难	主流解法是单调队列（deque），堆解法需「延迟删除」过期元素
480	滑动窗口的中位数	困难	双堆 + 延迟删除；或 SortedList——是 295 的滑窗版

面试常踩

1. Top-K 的方向反着用——找「前 K 大」用最小堆、找「前 K 小」用最大堆。背反了会写出错误且丑陋的代码。
2. Python 没有原生大根堆——只能值取负或用 (-key, value) 元组；用完别忘了取负还原。
3. 元组比较时 key 要放第一位——(freq, num) 比较先看 freq；如果反过来 (num, freq) 就成了按 num 排序。
4. heappush + heappop vs heappushpop——后者是单次 sift，常数小一倍。Top-K 模板里非常常用。
5. heapify 是 O(n)，但循环里反复 heapify 会变成 O(n²)——增量插入用 heappush，一次性建堆用 heapify。