处理动态数据里的“随时取最大最小”这类需求我最先想到的数据结构永远是堆。很多人对堆的第一印象是“堆排序”觉得它只是众多排序算法里的一种但实际用下来你会发现堆真正的价值是在数据不断增删的过程中始终能以很小的代价知道当前最大或最小的是谁。实时任务调度里新任务随时来优先级高的要插队先执行排行榜场景里热点数据一直变还要随时给出前 N 名日志系统里多个有序文件要合并成一个整体有序流。这些需求如果每次都全排序性能和实现复杂度都撑不住。这篇文章我会从堆的定义、数组存储、底层上浮下沉操作讲到 Python 官方 heapq 的使用边界、最大堆的实现方式再落到 Top K、K 路归并、任务调度三个真实场景最后整理几个我在实际写代码时踩过的坑。适合刚学数据结构的读者也适合准备算法面试或者写工具时需要优先队列的工程师。1. 为什么“随时取最大最小”这个问题值得单独学一种结构1.1 三种“找极值”方案的复杂度账在讲堆之前先把问题定义清楚。假设有一个不断变化的数据集我们要反复做两件事插入一个新元素取出当前最小的那个元素。如果数据只有十几个元素用列表直接存着每次 min() 扫一遍最省事但数据量到十万、百万级每取一次最小值都全列表扫描一遍就是不折不扣的灾难。我印象很深的一个场景有个内部系统需要实时统计每个秒级窗口内的最低报价每秒都有新报价进来还有过期报价要淘汰。最初用列表加 min()数据量一上去耗时肉眼可见地涨最后只能改成堆结构才把延迟压下来。把几种“动态取极值”的方案放在一起对比就能看得很清楚方案插入元素取最小值删除最小值一句话点评线性扫描O(1)O(n)O(n)数据量小时最简单直接每次重新排序O(n log n)O(1)O(1)排序成本太高数据一变就全乱二叉堆O(log n)O(1)O(log n)三项指标都很均衡有序数组O(n)O(1)O(1)插入时要移动元素代价太高平衡二叉搜索树O(log n)O(log n)O(log n)能实现但实现复杂度高得多从表里能明显看出堆其实是“插入、取极值、删极值”三者之间性价比最高的平衡点。实际工程里如果只是全量数据求一次最大最小值直接用 min() 和 max() 就够了但只要是“数据动态变化 反复取极值”的组合堆就是标准答案。1.2 堆的定义完全二叉树加堆序性堆并不是 Python 里的某个特殊类而是一种组织数据的方式。逻辑上它是一棵完全二叉树物理上它就是一个普通数组。所谓“完全二叉树”简单理解就是一层一层从左到右排节点排满一层才排下一层最后一层也必须尽量靠左。这个特性让它可以只用数组存不需要任何指针。堆序性则是第二个关键约束在小根堆里任何一个父节点都必须小于等于它的两个子节点所以整棵树的根节点必然是全局最小值。物理数组: [1, 3, 2, 7, 6, 4, 5] 逻辑结构: 1 / \ 3 2 / \ / \ 7 6 4 5看这个例子数组里 3 排在 2 前面但 3 是 1 的左孩子2 是 1 的右孩子它俩之间谁大谁小没有关系。堆只约束父子之间的顺序不约束兄弟之间的顺序。这也是为什么堆不是“已排好序的数组”而是一个有部分有序约束的结构。1.3 为什么优先队列经常和堆画等号优先队列是一个抽象概念它对外只承诺“插入元素”和“取出当前优先级最高或最低的元素”并不承诺先进先出。堆则是这个概念最常用、最简单的一种实现方式。Python 的 heapq 就是二叉堆实现所以很多文档干脆把两者混着说。普通队列是先进先出优先队列是按优先级出队。生活里最典型的例子是医院急诊的分诊台病情重的先处理不是先到先看。机场登机口让头等舱优先登机也是一种按优先级出队的逻辑。这里可以引出一个挺重要的结论会用堆就等于会用优先队列会写优先队列那它八成就是把堆包了一层。后面讲 heapq 的 API 和实战时你会反复体会到这句话。2. 数组里的完全二叉树下标映射与上浮下沉的手工实现2.1 用数组存完全二叉树下标公式如果根节点放在下标 0那么堆里任意一个元素的下标关系非常简洁下标 i 的父节点是(i - 1) // 2下标 i 的左孩子是2 * i 1下标 i 的右孩子是2 * i 2这三个公式就是堆里的全部“指针”。比如数组[1, 3, 2, 7, 6, 4, 5]下标 2 的元素是 2它的左孩子是下标 5 的 4右孩子是下标 6 的 5正好对应之前画的那棵树。这种存储方式最大的好处是内存连续、没有指针开销、缓存命中率高。在 Python 里底层就是 listappend 和按下标访问都是 O(1)。堆的一切操作最终都可以转化成数组下标运算这也是它实现起来特别简洁的原因。手动实现堆时永远要守住一个不变量对任意下标 i只要父节点存在就要求heap[parent] heap[i]。这个条件一旦被破坏堆就不再合法所有依赖“堆顶是最小值”的结论都会失效。2.2 上浮sift up新元素加入的自我修正向堆里插入一个新元素实现上分两步先把新元素追加到数组末尾这一步保持了完全二叉树的形态然后从末尾开始不断和父节点比较如果自己比父节点小就交换位置一路向上直到找到合适的位置。这个过程叫上浮。def sift_up(heap, i): while i 0: parent (i - 1) // 2 if heap[i] heap[parent]: heap[i], heap[parent] heap[parent], heap[i] i parent else: break时间复杂度是 O(log n)因为完全二叉树的高度就是 log n 量级每次上浮最多走这么多步。一个容易忽略的细节如果新元素已经比父节点大上浮会立即停止。堆不要求兄弟节点之间有大小关系所以“某个节点比它的兄弟大很多”是完全合法的。2.3 下沉sift down弹出堆顶后的重建弹出堆顶元素时如果直接把根节点删掉剩下的子树就散了。标准做法是记录堆顶它就是要返回的最小值。把数组最后一个元素搬到堆顶位置数组长度减一。从堆顶开始不断和左右孩子中较小的那个比较如果自己比孩子大就交换一路向下。为什么必须选“较小的孩子”交换因为堆序性要求父节点小于等于所有孩子交换后新父节点必须同时小于等于两个孩子。只有拿较小的孩子交换才能保证交换后另一个孩子依然满足条件。def sift_down(heap, i): n len(heap) while True: left 2 * i 1 right 2 * i 2 smallest i if left n and heap[left] heap[smallest]: smallest left if right n and heap[right] heap[smallest]: smallest right if smallest i: break heap[i], heap[smallest] heap[smallest], heap[i] i smallest手动实现建议把这两个函数各写几遍写到能顺手敲出来。因为 heapq 虽然已经封装好了但有些面试题会要求不能导入 heapq 的情况下手写堆或者在 LeetCode 上做“数组中第 K 个最大元素”这类题时背后其实就是这两个函数在运转。2.4 为什么 heapify 是 O(n)而逐个 push 是 O(n log n)这是学堆时最容易产生疑问的地方同样是把 n 个元素建成堆为什么heapq.heapify()是 O(n)而循环heappush是 O(n log n)关键区别在于 heapify 不是把每个元素都从根开始下沉而是从数组最后一个非叶子节点开始从下往上逐个做下沉调整。叶子节点完全不用调整接近叶子的节点下沉几步就够了只有靠近根的少数节点才需要下沉很多步。如果画一棵满二叉树从底层向上看倒数第二层节点数量大约是总数量的四分之一每个最多下沉 1 步倒数第三层数量更少每个最多下沉 2 步越往上层节点越少。把每一层的工作量加起来结果是一个收敛的级数总耗时是 O(n)。而逐个 heappush 等价于每个元素都从叶子位置向上浮 log n 层n 个元素加起来自然是 O(n log n)。所以日常建堆一定用heapq.heapify()不要写循环挨个 push。3. 吃透 heapq 八个核心接口才算会用优先队列Python 标准库里的 heapq 已经把上浮下沉全部封装好了。我不建议一开始就死磕源码里的私有函数但八个公开接口的行为和边界必须搞清楚否则很容易在细节上翻车。3.1 最基础的三板斧heapify、heappush、heappop先看 heapify它把一个普通列表原地变成合法堆import heapq nums [3, 1, 4, 1, 5, 9, 2, 6] heapq.heapify(nums) print(nums) # [1, 1, 2, 6, 5, 9, 4, 3]注意heapify 之后 nums 并不是严格从小到大排好序的。它只保证父节点小于等于子节点所以打印结果看起来有点“乱”。你要养成一个习惯堆不等于排序数组它只是一个能够快速取到最小值的结构。heappush 和 heappop 的用法非常直白heap [] for x in [5, 2, 8, 1]: heapq.heappush(heap, x) print(heap[0]) # 1堆顶就是当前最小值 print(heapq.heappop(heap)) # 1弹出并返回最小值 print(heap[0]) # 2有一个小知识点只要入堆的元素本身支持比较heapq 就能正常工作。int、float、str、tuple 都可以但 list、dict 这类不支持比较的类型直接入堆会报错。3.2 heapreplace 与 heappushpop一对容易搞反的兄弟这两个接口都做了同一件事往堆里塞一个新元素同时弹出一个元素。区别在于先后顺序和边界情况用错会出现让人摸不着头脑的结果。heapreplace(heap, item)先弹出堆顶再放入新元素。它要求堆非空。注意一个反直觉的点如果新元素比当前堆顶还小新元素也会进堆反而把原来的堆顶弹出去。heap [1, 3, 5] ret heapq.heapreplace(heap, 0) print(ret) # 1 print(heap) # [0, 3, 5] 新元素 0 反而进了堆heappushpop(heap, item)则是先做插入和弹出的整体判断再返回“两者中更小的那个”。如果新元素比堆顶还小它根本不会进堆直接返回新元素本身。heap2 [1, 3, 5] ret2 heapq.heappushpop(heap2, 0) print(ret2) # 0 print(heap2) # [1, 3, 5] 新元素不够格进都没进所以在 Top K 场景里正确的姿势通常是只有当前元素比堆顶大时才调用 heapreplace如果比堆顶小直接跳过不调用任何接口。3.3 nlargest / nsmallestK 远小于 N 时的高效方案如果只是要取前 K 大或前 K 小的几个元素heapq 内置的 nlargest 和 nsmallest 可以直接用nums [3, 1, 4, 1, 5, 9, 2, 6] print(heapq.nlargest(3, nums)) # [9, 6, 5] print(heapq.nsmallest(3, nums)) # [1, 1, 2]这两个函数内部是自适应的K 比较小时维护一个大小为 K 的堆K 接近 N 时直接排序反而更快它们会退化成排序处理。所以业务代码里取 TopK直接调它们是最稳的选择。nlargest 还支持 key 参数比如按字典里的字段取前几users [{name: a, age: 20}, {name: b, age: 35}, {name: c, age: 28}] top heapq.nlargest(2, users, keylambda u: u[age])另一个值得记住的接口是 heapq.merge它能把多个已经有序的序列惰性合并成一个有序迭代器底层也用到了堆。后面讲 K 路归并时我还会提到。3.4 最大堆的三种实现姿势Python 的 heapq 只支持小根堆“堆顶最小”。想要实现最大堆常见有三条路。第一种直接存负值适用于 int 和 floatmax_heap [] heapq.heappush(max_heap, -5) heapq.heappush(max_heap, -2) heapq.heappush(max_heap, -8) cur_max -heapq.heappop(max_heap) # 5第二种包装自定义对象重写__lt__。heapq 比较元素时主要调用所以只要让“数值大的”被视为“更小”即可class MaxHeapItem: def __init__(self, val): self.val val def __lt__(self, other): return self.val other.val第三种在元组里手动反转权重(-priority, counter, data)。这种方式既能处理数值优先级又能在优先级相同时靠 counter 避免比较到不可比较的元素。需要提醒一下取负值法虽然对字符串等类型不适用但绝大多数“动态取最大”的场景配合小根堆和负数已经够用了不一定要额外造一个最大堆类。4. 三个高频实战Top K、K路归并与优先级任务调度4.1 Top K用最小堆维护“最大的 K 个”先想清楚一个反直觉的点为什么求最大的 K 个元素反而要用最小堆假设有一个大小为 K 的“候选区”我希望它始终保存当前遇到的最大的 K 个元素。判断一个新元素能不能进入候选区只需要和候选区里最小的那个比如果新元素比它大就把这个最小的踢出去新元素进来如果新元素连候选区最小的都比不过那它肯定没资格进前 K。候选区里最小的那个正好就是堆顶。所以最小堆在这里是“淘汰线”的守卫。实现代码如下import heapq def top_k_largest(nums, k): if k 0: return [] heap [] for x in nums: if len(heap) k: heapq.heappush(heap, x) elif x heap[0]: heapq.heapreplace(heap, x) return heap复杂度是 O(n log k)。当 K 远小于 N 时它明显优于全排序 O(n log n)而且内存只需要 O(K)。这道题的变形在面试里出现频率非常高建议把“为什么用最小堆求最大 K 个”练到能脱口而出。4.2 合并 K 个有序序列多路归并的堆解法场景是这样的手里有 K 个已经各自有序的数组或文件想把它们合并成一个整体有序的大序列。最笨的办法是把所有元素塞进一个列表再排序但如果面对的是 K 个几个 GB 级别的日志文件这个方案内存会直接爆掉。堆的做法非常精妙把每个序列的第一个元素放进堆用三元组记录值来自哪个序列位置是多少。弹出堆顶它一定是当前全局最小的一个写入结果。从弹出的那个序列取下一个元素继续放进堆。重复直到所有序列耗尽。import heapq def merge_k_sorted_arrays(arrays): heap [] for arr_idx, arr in enumerate(arrays): if arr: heapq.heappush(heap, (arr[0], arr_idx, 0)) result [] while heap: val, arr_idx, pos heapq.heappop(heap) result.append(val) if pos 1 len(arrays[arr_idx]): nxt arrays[arr_idx][pos 1] heapq.heappush(heap, (nxt, arr_idx, pos 1)) return result注意元组里为什么要带 arr_idx 和 pos当两个值相等时heapq 会继续比较后面的字段arr_idx 和 pos 都是整数能保证比较不崩。更极致一点如果每个序列特别长完全不需要一次性读进内存可以把序列换成文件行迭代器堆里只保存每个文件的当前行。这就是外部归并排序的雏形。实际项目里如果序列本身就是有序迭代器直接heapq.merge(*iterators)能替代手写版本但我依然建议把这段逻辑手写一遍因为很多算法题不会直接告诉你“这是 K 路归并”。4.3 优先级任务调度为什么元组里要带序号任务调度的需求很直白任务带着优先级来优先级高的先被处理。用堆实现一个优先队列非常简单import heapq from itertools import count class PriorityQueue: def __init__(self): self._heap [] self._counter count() def put(self, priority, item): heapq.heappush(self._heap, (priority, next(self._counter), item)) def get(self): priority, _, item heapq.heappop(self._heap) return priority, item这里有一个从实际运行中踩出来的坑如果直接存二元组(priority, item)当两个任务优先级相同时heapq 会比较第二个元素 item。如果 item 是字符串还勉强能比如果是自定义对象或者 dict直接 TypeError。解决办法就是这里的“带序号三元组”priority 相同时比较 counter而 counter 永远不会相同比较立刻停止根本轮不到 item。counter 同时还保证了同优先级任务之间的相对顺序也就是 FIFO。4.4 堆在 Dijkstra、爬虫队列等场景里的延伸Dijkstra 最短路径算法里每次需要从未访问节点中挑出距离最小的节点。朴素实现是 O(V) 扫描整体复杂度是 O(V²)。改成优先队列后这一步降为 O(log V)图中节点多时效果非常明显。爬虫调度里如果希望优先抓取某个权重更高的 URL用的也是同一个堆结构。消息中间件里的延迟队列本质上是按到期时间排队到期时间就是优先级。理解了堆和优先队列的关系再看这些场景基本都是同一个模式换了个马甲的动态极值问题。5. 堆的高频踩坑与性能调优记录5.1 可变对象入堆后修改字段堆序就崩了heapq 只保证你在 push 和 pop 操作时维护堆序不保证对象入堆之后不会被外部改动。如果入堆对象是可变对象入堆后你改了它的优先级字段堆就会“失效”。class Task: def __init__(self, priority, name): self.priority priority self.name name def __lt__(self, other): return self.priority other.priority t Task(5, a) heap [] heapq.heappush(heap, t) t.priority 1 # 堆序被破坏heap[0] 不再是正确的最小值提示堆只是保证“操作时”的不变量它不是一份会自动感知外部修改的动态数据结构。如果确实有“任务优先级会动态变化”的需求常规做法是不修改已入堆对象而是重新 push 一个带有新优先级和版本号的对象旧对象保留在堆里但标记失效弹出时检查标记并跳过。这种“惰性删除”在很多语言里都常见代价是堆里可能积累一些无效对象需要定期清理。5.2 优先级相同的二元组会触发比较器 TypeError直接看这段代码import heapq h [] heapq.heappush(h, (2, {a: 1})) heapq.heappush(h, (2, {b: 2})) # heapq.heappop(h) # 这里大概率 TypeError第二个元组入堆时heapq 发现第一个元素都是 2就会去比较第二个元素dict 之间没法比较大小于是抛异常。解决办法上节已经给过了用带序号的三元组。这也解释了为什么很多工程代码里的优先队列元素永远是“数字元组”而不是裸对象。5.3 线程安全heapq 不是为并发设计的heapq 操作的是普通 list内部没有任何锁。多线程并发 push 和 pop 时可能同时修改列表结构轻则结果错乱重则列表直接损坏。如果是生产者消费者模型直接用queue.PriorityQueue更合适。它内部就是对 heapq 加了线程锁并提供了阻塞式 put 和 get队列满了会阻塞写入队列空了会阻塞读取非常适合任务分发场景。如果只是单线程或者已经自己在外部用锁保护了那就不需要引入 PriorityQueue因为锁是有开销的裸 heapq 在这种场景下更快、更可控。5.4 性能优化笔记比较开销、建堆方式与内存规模几个实际调优方向对性能影响很大。第一减少比较开销。堆操作的核心是堆序比较比较越重性能越差。与其把整个大对象塞进堆不如在堆里只放“比较键 引用或索引”比如(value, id)每次比较只做整数比较。第二建堆用 heapify。前面已经反复强调过n 越大区别越明显。100 万元素时heapify 比循环 heappush 能快一个量级。第三取前 K 直接用 nlargest 和 nsmallest。它们已经针对 K 的大小做了路径选择比自己手写堆更稳。第四注意 Python 对象的内存开销。堆里的每个元素都是独立 Python 对象百万级元素时list 本身是指针数组对象又有独立对象头内存很容易到几十 MB。如果数据是纯数值且内存敏感可以用 array 模块存数组再基于下标自己维护堆逻辑不过这是性能特化场景常规开发不需要。回想一下堆的代码量其实非常小核心操作就两个上浮和下沉。它之所以难是因为它的所有正确性都建立在一个不变量上父节点永远小于等于子节点。只要这个不变量被维护住heapq 的每个接口都值得信赖一旦被破坏所有结论都会失效。这也是为什么我现在每次用堆都会在心里默念一遍入堆出堆只有一条规则但这条规则必须被严格执行。
