前阵子帮一个做日志分析的同事改代码，他那段程序要从每天上亿条请求日志里捞出响应时间最长的100条。第一版实现特别直白：全量解析完排个序，再切片取前100。结果呢？近一亿条记录解析完直接吃掉16G内存，光排序就跑了40多分钟。后来我给他换成了Python数据结构里的堆——准确说是标准库的heapq模块，内存占用瞬间压到几十MB，时间也掉到分钟级以内。今天这篇博文，想把Python里堆的用法、底层原理和真实业务里那些坑一次讲透。适合正在啃数据结构与算法、准备面试刷题的同学，也适合工作中经常被TopK、优先队列这类需求缠住的工程师。
很多人对堆的第一印象是"二叉树"，但真正在Python里用起来，它其实就是一个列表，加上heapq提供的一组操作函数。这个"看起来是树、存起来是数组"的结构，让无数排序、调度、流式计算问题变得极其优雅。我下面会从实战场景切入，一路拆到源码级别，最后把那些文档里不会写的坑全部摊开。
1. 面试官问我TopK：为什么排序挂了而堆活了下来
1.1 一个被排序算法坑惨的真实场景
先说回开头那个日志分析需求。假设你有1000万条记录，要取耗时最长的Top 100。新手最自然的想法是：
records = load_all_records() # 千万级 records.sort(key=lambda x: x.cost, reverse=True) top100 = records[:100]这段逻辑没错，但它有俩致命问题：一是load_all_records()必须把全部数据加载进内存，千万条记录自带字段一大堆，内存直接爆炸；二是sort()是全局排序，复杂度O(n log n)，但你只需要100个结果，剩下99999900条记录的排序工作全是浪费。
堆的思路完全不同：维护一个容量只有100的小根堆，遍历数据流，只要当前元素比堆顶大，就把堆顶弹出去、把新元素压进来。这样全程内存占用只有100个元素，时间复杂度O(n log k)，k是堆的大小。n从千万级降到100的量级，内存和CPU双丰收。
这就是我要说的第一件事：堆本质上是"只关心局部最优"的数据结构。它不追求把所有元素排好序，只要求能快速拿到当前最值，并且代价是O(log n)的插入删除。排序是慢工出细活，堆是快刀切乱麻，场景不同，没有谁绝对更好。
1.2 堆的直觉理解：一棵"偏心"的完全二叉树
想用好堆，得先建立直觉模型。堆是一棵完全二叉树，所谓完全，就是除了最后一层，上面每层都是满的，最后一层的节点从左往右紧密排列。
小根堆额外满足一条规则：任意父节点的值都不大于它的子节点。这意味着根节点永远是整棵树的最小值，这就是堆能O(1)取最值的原因。
你可能会问：完全二叉树为什么要存在数组里？因为完全二叉树没有空洞，可以精确地按下标铺开，不需要额外的指针。如果你把堆存在列表里，下标关系是固定的：
- 下标 i 的父节点是
(i - 1) // 2 - 左孩子是
2 * i + 1 - 右孩子是
2 * i + 2
每次插入新元素，先放到数组末尾，然后一路和父节点比较，小就往上换，这个过程叫"上浮"；删除堆顶时，把数组末尾的元素挪到堆顶，然后一路下沉，这个过程叫"下沉"。上浮下沉都只走树的高度这么多步，高度是O(log n)，所以堆的插入删除都是O(log n)。
这里我插一句个人体会：很多人学堆卡在"数组里怎么看出树形结构"上，我的建议是自己手写一遍下标推演。比如列表[1, 3, 5, 9, 7, 8]，下标2的值5，父节点是下标0的1，左孩子是下标4的7，右孩子下标5的8——画出来就是一棵标准小根堆。有了这个肌肉记忆，看heapq源码会轻松得多。
2. heapq源码级拆解：一棵隐身在列表里的树
2.1 五个核心API，够用一整年
Python标准库的heapq模块，把堆的操作封装成了几个函数。我把它当工具，但常年只用这五个：
| 函数 | 作用 | 复杂度 |
|---|---|---|
heappush(heap, item) | 将一个元素压入堆，自动调整到合适位置 | O(log n) |
heappop(heap) | 弹出堆顶（最小值），并调整结构 | O(log n) |
heapify(x) | 原地将一个普通列表整理成堆 | O(n) |
heapreplace(heap, item) | 先弹出堆顶，再压入新元素 | O(log n) |
nlargest(n, iterable) | 从一个可迭代对象里取最大的n个 | 视n大小而定 |
其中heappush和heappop是地基，剩下的是在它们上面做了组合优化。
举个例子，如果你需要"取最大值、删最大值，同时还要往里加新元素"，标准做法是heappop再heappush，但这样要两次O(log n)操作。用heapreplace一次就干完，而且它是先弹后压，底层操作更短，实测在频繁更新场景下能快20%到30%。
2.2 上浮与下沉：源码里藏着的高效细节
heapq的源码不算长，核心内部函数是_siftdown和_siftup。
_siftdown负责"上浮"：从指定位置开始，不断把当前节点和父节点比较，如果当前节点更小就和父节点交换位置，直到到达根或不再小于父节点。它用在heappush和heapify的部分环节。
_siftup则不是简单地把堆顶元素一路和小孩子比下去，而是做了个优化：先把最小的那个孩子提上来，形成一条从堆顶到某个叶子节点的空洞路径，最后把堆尾元素放进空洞，再反向做一次上浮。这个设计能减少元素交换的次数。
这里我多说一句：源码里_siftup配合_siftdown一起用，很多学习资料都没讲清楚。简单记结论就行：CPython用C实现大部分逻辑，heapq模块本身是Python写的，所以你能直接读源码——读一遍，你对堆的理解会超过90%只调API的人。
2.3 heapify为什么是O(n)：矮树省出来的复杂度
很多人背过"建堆复杂度O(n)"，但不知道原因。我用大白话解释：
如果逐个往空堆里插入n个元素，每次插入O(log n)，总复杂度O(n log n)。但heapify不是这样——它拿到一个乱序列表，从最后一个非叶子节点开始，往前逐个做下沉调整。
关键点在于：大部分节点位于树的底部，底部节点下沉需要走的距离很短。一棵完全二叉树里，叶子节点占了一半，它们根本不下沉；倒数第二层的节点最多下沉一层，倒数第三层最多下沉两层……把每层的下沉次数按高度加权求和，结果是收敛的，所以总工作量是O(n)。
这个结论直接用，不用记推导过程：面对一个已有列表，建堆用heapify，别用循环heappush。实测100万个随机数，heapify只花0.1秒级别，循环heappush要好几秒，差距几十倍。
3. 大根堆的三种实现，Python的"逆向"思维
3.1 取负法：最实用但容易踩坑
Python的heapq只有小根堆，可很多场景要的是"最大值优先"，比如按评分最高的用户、按最紧急的过期时间。第一种最粗暴的方案是存负值。
import heapq big_heap = [] data = [3, 1, 4, 1, 5, 9, 2, 6] for x in data: heapq.heappush(big_heap, -x) # 降序取最大值 while big_heap: print(-heapq.heappop(big_heap), end=" ") # 9 6 5 4 3 2 1 1思路一句话：小根堆里最"小"的负数，就是原数据里最大的数。取出来再取负，就还原了。
这个方案简单直接，但有两个坑必须注意：
第一个坑是取负后的数值范围问题。如果你处理的是无符号整数或者极大值，取负可能导致溢出或精度问题。Python整数是任意精度没这问题，但要提防浮点数，-0.0和0.0在比较上相等，可能打乱预期。
第二个坑更隐蔽：存入的如果是元组，取负只能作用于第一个元素，如果你希望按元组的多个字段排序，取负会破坏后续字段的顺序。比如(-score, name)里面name是按正序存的，你要"同分时名字倒序"，就得再想别的办法。
3.2 自定义对象的__lt__：治本之策
更稳妥的做法是定义自己的数据类，重写__lt__。heapq比较元素时用的是"小于"，只要你的对象能回答a < b，它就能进堆。
class Task: def __init__(self, priority, name): self.priority = priority self.name = name def __lt__(self, other): # 注意：想让优先级大的先出，就把比较反过来 return self.priority > other.priority def __repr__(self): return f"Task({self.priority}, {self.name})" tasks = [Task(3, "低"), Task(10, "高"), Task(7, "中")] heapq.heapify(tasks) heapq.heappop(tasks) # Task(10, 高)这里有个文档不会告诉你的细节：heapq内部用的是<和>比较，所以你只重写__lt__就够了。但默认的__eq__也会参与比较，如果两个对象优先级相同、没有实现__lt__，代码会崩溃，提示TypeError: '<' not supported between instances。建议同时实现__eq__，或者干脆在__lt__里处理平级情况。
3.3 元组多字段时的反直觉问题
实际业务里经常要按"优先级+时间"排序，很多人直接存(priority, timestamp, data)。这里隐藏一个坑：如果两个元素的priority相同，heapq会继续比较timestamp，如果timestamp类型不一致，直接TypeError。而且次数多了以后，堆里会积压大量同优先级旧数据，先入先出的语义得不到保证。
我的经验是：在业务堆里显式加入一个自增序号作为第二排序字段，保证严格有序。
import itertools seq = itertools.count() heap = [] heapq.heappush(heap, (priority, next(seq), data))为什么？因为heapq要求堆内元素必须可以互相比较，一旦比较不出来，整个堆就崩了。自增序号保证了任何两个元素都有确定的先后顺序，这是工程上非常划算的保险。
4. 堆在真实业务里的三个落地场景
4.1 大文件TopK：一次遍历，内存不破防
回到开头那个日志问题。正确做法是流式处理文件，一行一行读，堆内始终只留K个最大元素：
import heapq def top_k_from_file(file_path, k): heap = [] with open(file_path, 'r', encoding='utf-8') as f: for line in f: cost = extract_cost(line) # 解析出耗时 if len(heap) < k: heapq.heappush(heap, cost) elif cost > heap[0]: heapq.heapreplace(heap, cost) return heap这里有两个要点。第一，heap[0]是堆顶，也就是当前K个元素里最小的那个，新元素比它大才值得替换；第二，用heapreplace而不是先heappop再heappush，省一次操作。实测解析1亿行日志，这个程序的内存占用稳定在K*16字节左右，K=100时几乎可以忽略不计。
4.2 定时器与延迟任务：优先队列的正确玩法
实现一个简单的延迟任务调度器，堆是最合适的结构。把任务的触发时间戳作为堆排序依据，每次循环只需要看堆顶，如果时间到了就弹出执行：
import heapq import time class TimerScheduler: def __init__(self): self._queue = [] self._seq = itertools.count() def add(self, delay, func): heapq.heappush(self._queue, (time.time() + delay, next(self._seq), func)) def run(self): while self._queue: due_time, _, func = self._queue[0] now = time.time() if now < due_time: time.sleep(due_time - now) heapq.heappop(self._queue) func()这种实现的优点是：新增一个任务只需O(log n)，主循环永远只检查堆顶，不需要扫描全部任务。很多消息队列组件里的延迟队列就是类似思路。小细节：因为两个任务可能触发时间完全一样，我加了next(self._seq)作为第二排序字段，保证不会出现比较异常。
4.3 双堆维护数据流中位数
如果数据源源不断进来，想随时拿到当前所有数据的中位数，堆能给出惊艳的方案：用一个最大堆存左半部分，一个最小堆存右半部分，两堆数量差不超过1。中位数就是堆顶之一或它们的平均值。
Python没有内置大根堆，所以左半部分用取负法：
class MedianFinder: def __init__(self): self.left = [] # 最大堆，存负值 self.right = [] # 最小堆 self.median = None def add_num(self, num): if not self.left or num <= -self.left[0]: heapq.heappush(self.left, -num) if len(self.left) > len(self.right) + 1: heapq.heappush(self.right, -heapq.heappop(self.left)) else: heapq.heappush(self.right, num) if len(self.right) > len(self.left): heapq.heappush(self.left, -heapq.heappop(self.right)) if len(self.left) > len(self.right): self.median = -self.left[0] elif len(self.left) < len(self.right): self.median = self.right[0] else: self.median = (-self.left[0] + self.right[0]) / 2这套结构插入O(log n)，查询中位数O(1)。我当年面试遇到这题时，现场手写大概花了十分钟，但真正在流式监控这种场景里用起来，你会觉得这十分钟写得太值了。
5. 别把内存里的"堆"和数据结构里的"堆"搞混
5.1 两个"堆"：一个是结构，一个是地盘
搜索热词里大量出现"堆和栈""堆外内存""进程堆大小8000报OOM"，这里必须做个彻底区分。
数据结构里的堆，是我们前面讲的完全二叉树，解决问题的，是逻辑结构。
JVM或者操作系统内存模型里的堆，是一块内存区域，存放对象的，是运行时的内存分配策略。两者的英文都是heap，但完全不是一回事。
在Python语境里，事情的画风又不一样：Python所有对象都分配在堆上。这意味着你写a = [1, 2, 3]，列表对象本体在堆区，变量a只是个指向它的引用。Python的"栈"主要用来存放函数调用帧（局部变量、返回值等），所以递归过深时你遇到的是RecursionError，本质是调用栈耗尽了，不是堆的问题。
5.2 热词里那些OOM和堆栈溢出，到底在报什么
下面几个高频搜索词，我逐一给你翻译成人话：
| 搜索词 | 实际情况 | 跟heapq的关系 |
|---|---|---|
| 进程堆大小调整为8000还是报OOM | 这是JVM堆内存（-Xmx）不够，或程序有内存泄漏 | 无关 |
java.lang.OutOfMemoryError | JVM堆区无法分配新对象 | 无关 |
| win11堆栈区溢出 | 一般是递归过深或超大局部变量，调用栈溢出 | 无关，但要警惕Python的RecursionError |
| 堆外内存 | 绕过堆管理直接使用本地内存，比如NIO的DirectBuffer | 无关 |
| 编译器的堆空间不足 | IDE/构建工具的JVM堆不够 | 无关 |
我见过不少刚学数据结构的人，搜"堆"却总是搜到JVM调优帖子，以为heapq能解决内存OOM，那真是误会大了。heapq是逻辑数据结构，只解决排序和取最值问题，不帮你申请内存。
至于Python这边的"栈溢出"，主要就是递归没写好。比如快速排序的递归实现，数据量一大就可能撞上Python默认的递归上限1000，报RecursionError。解决办法要么改迭代，要么调sys.setrecursionlimit()，但调递归上限是治标不治本，真正复杂的递归场景建议直接上迭代栈。
5.3 Python的"堆"在哪：对象分配的小知识点
Python对象默认就在堆区，但你写脚本时基本感知不到它。只有做内存分析时，才会用tracemalloc或psutil看到进程RSS涨跌。如果你想确认某个对象占了多少内存，可以用：
import sys print(sys.getsizeof([1, 2, 3])) # 例：80字节左右这个getsizeof只统计对象本身，不含引用对象的内部元素，做深层次内存分析还得用pympler之类的库。这些小工具解决的是"内存去哪了"的问题，跟用堆做TopK是两码事。
6. 使用heapq常见的五个坑与性能实测
6.1 坑一：heapify是原地操作，它不返回新堆
这是新手最容易摔的一跤：
import heapq data = [3, 1, 4, 1, 5] result = heapq.heapify(data) print(result) # None print(data) # [1, 1, 4, 3, 5] 数据已经被改heapify返回None，所有调整都在原列表上完成。你要是写成data = heapq.heapify(data)，data直接变None。解决办法就是别赋值，直接调。
6.2 坑二：堆内的元素必须是可比较的
heapq把元素当"能互相比大小"的东西。如果你塞进去的是不同类对象，或者自定义类没实现__lt__，比较时直接抛TypeError。
heap = [] try: heapq.heappush(heap, ("a", 1)) heapq.heappush(heap, (2, "b")) # str和int比较，崩 except TypeError as e: print(e)解决思路我已经在前面反复说过了：要么统一元素类型，要么加自增序号兜底。尤其在存元组时，第二字段的类型一致性特别容易被忽视。
6.3 坑三：修改堆内元素后，堆序不会自动恢复
heapq没有提供"更新某个元素"的API。我踩过这个坑：把某个任务的优先级调低后，直接改了元组里的字段，结果后面取出来的根本不是最小元素，逻辑全乱了。
正确做法有三种：
- 先找到目标元素移除，再重新
heappush。缺点是O(n)查找。 - 标记法：额外维护一个
removed = set()，取出堆顶时如果已删除就跳过，更新时直接压入新元素。适合延迟队列这种低频更新场景。 - 数据量小就直接重建堆，反正
heapify是O(n)，比纠结精细更新省心。
6.4 性能实测：堆到底比排序快多少
我在一台普通笔记本上做了一组简单测试，数据是100万个0到1亿之间的随机整数，取Top 100：
| 方案 | 耗时 |
|---|---|
sorted(data, reverse=True)[:100] | 约1.2秒 |
heapq.nlargest(100, data) | 约0.18秒 |
| 手动维护堆（for + heapreplace） | 约0.25秒 |
heapify+ 反复heappop（取全部100万） | 约1.8秒 |
有意思的是，CPython的nlargest内部做了优化：当n相对于序列长度较小时，它走堆路线；当n接近序列长度时，它反而改用sorted。所以绝大多数场景你直接用nlargest就行，不用自己造轮子。
但要注意：sorted(data)[:100]仍会生成一个100万元素的完整排序列表，内存占用高；nlargest内部也会创建一个长度为n的堆，内存开销小得多。
6.5 手写堆排序：十分钟检验你懂没懂
如果你读完前面还觉得手痒，建议自己写一遍堆排序。最干净的写法是：
import heapq def heapsort(iterable): h = list(iterable) heapq.heapify(h) return [heapq.heappop(h) for _ in range(len(h))]这段代码虽然只有三行，但走通它需要你明白：heapify建堆、每次heappop取最小、弹出的过程中剩余元素始终保持堆序。写完后，把heappop换成手写下沉逻辑，再写一遍，你基本就掌握堆了。
我在带新人时常让他们做这个练习，因为他们能自己把"数组下标树"和"上浮下沉"串起来，之后再看任何优先队列的代码都不怵。堆这个数据结构，学的时候觉得抽象，用起来是真香，尤其在大数据量 TopK、动态中位数、任务调度这些场景里，它几乎是不可替代的解法。
