你要是最近在准备算法面试数据流中位数和滑动窗口中位数这对双子星应该已经被各种高频题清单反复点名了。名义上是两个题本质其实是同一个模型在一个动态变化的集合里维护一个能随时回答“当前中位数是多少”的结构。而解决这个模型最经典、也最常被面试官期待听到的工具就是对顶堆。这篇文章我会用聊天的口吻把对顶堆的原理、数据流中位数的实现、滑动窗口中位数的实现以及我在面试和带人准备面试时踩过的一些坑全部串起来讲一遍。不管你是刚开始刷题的新手还是已经刷了一段时间但遇到堆就发怵的选手按这条线走一遍应该都能对这类题建立起自己的判断框架。1. 为什么数据流中位数和滑动窗口中位数都绕不开对顶堆1.1 中位数不是“排序后取中间”这么简单中位数的定义大家小学就学过把一组数按从小到大排好取中间那个。如果是偶数个数取中间两个的平均值。但这里藏着一个问题定义是静态的而数据流和滑动窗口里的集合是动态的。数据流意味着每来一个数集合规模就变大一点滑动窗口意味着每移动一格走一个旧数、进一个新数。如果每次求中位数都把整个集合重新排一遍复杂度就是 (O(n \log n)) 甚至更高。面试官真正想看的不是你能不能写出排序而是你能不能发现中位数本质上是一个“位置维护”问题我只关心中间那个位置上的数没必要把整个序列都排好。这就像你维护一个班级的成绩单如果只想知道谁排中间不需要每次都把全班重新按名次贴一遍墙只要把左右两边的人数控制住就行了。对顶堆就是干这件事的结构。1.2 对顶堆一个管小区一个管大区对顶堆的英文叫 two heaps中文翻译很形象。它由两个堆组成一个最大堆负责装数据中“比较小的一半”我们叫它左堆。一个最小堆负责装数据中“比较大的一半”我们叫它右堆。左堆的堆顶是左半边最大的元素右堆的堆顶是右半边最小的元素。只要保证左堆里所有元素都小于等于右堆里所有元素并且两个堆的大小相差不超过 1那么中位数就能直接用两个堆顶算出来如果总数是奇数左堆比右堆多一个左堆堆顶就是中位数。如果总数是偶数两个堆一样大中位数是左堆堆顶和右堆堆顶的平均值。为什么能保证左堆所有元素都小于右堆靠的不是每次插入后重新整理全堆而是靠插入时的“乾坤大挪移”先往某一边塞再取堆顶扔到另一边。堆本身只保证堆顶有序但经过这种转移后左右两半的相对顺序就会维持住。1.3 插入后的平衡口诀先塞左边再丢右边不够再捞很多初学者第一次写对顶堆时会把平衡逻辑搞得很复杂又是 if 又是 while。其实核心就三句话新数先塞进左堆最大堆。从左堆弹出堆顶丢到右堆最小堆。如果此时左堆比右堆少再从右堆弹出堆顶捞回左堆。第一步先把新数放到左堆让它和左堆里的元素比较一轮左堆的堆顶自然就是整个左半边最大元素。第二步把左堆堆顶丢到右堆实际上是在做“分流”如果新数很大它会通过左堆堆顶的位置被送进右堆如果新数很小被丢过去的是原来的左堆堆顶新数会留在左堆更靠里的位置。第三步调整大小保证左堆永远不比右堆少。这个流程每步都是 (O(\log n))但逻辑非常稳面试时不容易写乱。2. 数据流中位数双堆直落代码一次写对2.1 准备两个堆之前先想清楚大小约定写代码之前先把“堆的大小关系”定死。我最常用的约定是左堆大小 右堆大小或者左堆大小 右堆大小 1。也就是说奇数个数时中位数单独放在左堆堆顶。这个约定不是唯一的有人喜欢右堆多一个也有人喜欢总是在右堆取中位数。但你一定要固定一个并且写代码时每一步都对照这个约定。我见过太多候选人代码写得很顺但 heap 大小关系前后矛盾最后 findMedian 里返回了错误的一边。2.2 addNum 最稳的“三步走”写法用 Python 写堆有个小别扭标准库 heapq 只有最小堆没有最大堆。技巧是存相反数。比如原始值 5 存成 -5那么堆顶是最小的负值对应的就是最大的原始值。以 LeetCode 295 数据流的中位数为例核心代码可以这样写import heapq class MedianFinder: def __init__(self): self.left [] # 最大堆存负值 self.right [] # 最小堆存正值 def addNum(self, num: int) - None: # 第一步新数先进左堆 heapq.heappush(self.left, -num) # 第二步把左堆堆顶左半边最大值丢到右堆 heapq.heappush(self.right, -heapq.heappop(self.left)) # 第三步如果左堆比右堆少从右堆捞一个回来 if len(self.left) len(self.right): heapq.heappush(self.left, -heapq.heappop(self.right)) def findMedian(self) - float: if len(self.left) len(self.right): return (-self.left[0] self.right[0]) / 2.0 return float(-self.left[0])这段代码的每一步都对应前面说的“先塞左边再丢右边不够再捞”。你可能会问第一步直接塞左堆第二步又从右堆把左堆最大值丢回去是不是多此一举不是。如果不经过这一步新数到底属于左半边还是右半边你必须额外判断而经过“进左堆再弹出最大值丢右堆”这个判断被堆自动完成了。这是一种“以空间换逻辑简单”的做法非常推荐面试时用。2.3 findMedian 是小事但别写错返回类型findMedian 看起来就是取堆顶但有两个细节偶数个数时返回浮点数一定要除以 2.0不是除以 2。Python 里整数除法会直接截断。奇数个数时左堆堆顶存的是负值要转回原始值再返回。很多人在白板写代码时觉得这个题简单最后挂在返回类型这种小事上非常可惜。2.4 完整实现与复杂度复盘数据流中位数的时间复杂度操作复杂度说明addNum(O(\log n))最多几次堆的 push/popfindMedian(O(1))只需要看两个堆顶空间(O(n))所有数据都存放在两个堆里这个复杂度为什么好因为这意味着我们可以维护一个无限增长的数据流任何时候想知道中位数几乎是瞬间出结果。如果面试官追问“能不能更快”你要知道从理论上讲基于比较的插入结构最低就是 (O(\log n))所以对顶堆已经是最优解之一。3. 滑动窗口中位数最难的不是堆是删除3.1 为什么排序法在滑动窗口里很快就不行了如果说数据流中位数是对顶堆的入门那么滑动窗口中位数就是对顶堆的进阶考验。题目一般是这样给定一个数组 nums 和窗口大小 k窗口每次往右移动一位要求返回每个窗口的中位数。最朴素的做法是每次截取窗口内 k 个数排序取中位数。窗口移动 n 次每次排序 (O(k\log k))总复杂度 (O(n k \log k))。k 一旦上千这题基本就废了。更优的思路是沿用对顶堆把所有窗口内元素按大小分成左右两半堆顶提供中位数。窗口每次移动时做两件事删掉离开窗口的数加入新进来的数。插入我们已经会了真正的难点在“删除”。3.2 堆的删除痛点与“延迟删除”设计堆本身支持删除堆顶但不支持直接删除任意元素。你要是想删掉一个不在堆顶的数只能先标记它等它慢慢浮到堆顶再真正弹出。这个技术叫延迟删除英文一般叫 lazy deletion。具体做法是维护一个哈希表记录“已经被删除但还没从堆里真正弹出去的元素”以及它们的次数。比如窗口准备移动旧元素 x 要离开我不马上从堆里物理删除 x而是delayed[x] 1。等到某次查看堆顶时如果堆顶元素恰好是 x并且 delayed 里记着它就把它弹出去同时delayed[x] - 1。如果堆顶不是 x说明 x 还埋在堆里暂时不影响中位数计算那就留着。这个思路很像现实里的“延迟发货”订单先标记成取消但货物还在仓库里只有轮到这个货出库时才把它拦下来。它保证了堆的其他操作复杂度仍然是 (O(\log k))只是多了一个哈希表的常数开销。3.3 用 balance 维持双堆规模的诀窍数据流中位数可以用len(left) len(right)来平衡但滑动窗口里有延迟删除物理堆的大小可能和逻辑大小不一致。这时候再用 len 判断就不准了需要额外用一个变量 balance 记录左右堆的逻辑大小差。我约定 balance 左堆逻辑大小 - 右堆逻辑大小。目标仍然是 balance 等于 0 或 1。于是每个滑动步骤变成这样旧元素离开如果旧元素在左堆balance 减 1在右堆balance 加 1。新元素加入如果新元素应该进左堆balance 加 1进右堆balance 减 1。最后根据 balance 的正负往堆之间搬一个元素并且每搬一次balance 相应减 2 或加 2。为什么要加 2因为搬一个元素过去一边少了 1另一边多了 1两边差距直接变化 2。这个细节很容易写错面试时建议用一个小用例现场推一下。3.4 滑动窗口中位数的完整实现下面这份代码是 LeetCode 480 滑动窗口中位数的常见解法我加了比较详细的注释from typing import List import heapq from collections import defaultdict class Solution: def medianSlidingWindow(self, nums: List[int], k: int) - List[float]: small [] # 最大堆存负值 large [] # 最小堆存正值 delayed defaultdict(int) balance 0 ans [] def prune(heap, is_small): # 把堆顶已经标记为删除的元素真正弹出去 while heap: val -heap[0] if is_small else heap[0] if delayed[val] 0: delayed[val] - 1 heapq.heappop(heap) else: break def make_balance(): nonlocal balance if balance 0: # 左堆多了把左堆堆顶挪到右堆 prune(small, True) heapq.heappush(large, -heapq.heappop(small)) balance - 2 elif balance 0: # 右堆多了把右堆堆顶挪到左堆 prune(large, False) heapq.heappush(small, -heapq.heappop(large)) balance 2 # 初始化第一个窗口 small [-x for x in nums[:k]] heapq.heapify(small) large [] for _ in range(k // 2): heapq.heappush(large, -heapq.heappop(small)) balance len(small) - len(large) def get_median(): prune(small, True) prune(large, False) if k % 2 1: return float(-small[0]) return (-small[0] large[0]) / 2.0 if k 1: ans.append(get_median()) for right in range(k, len(nums)): remove_val nums[right - k] add_val nums[right] # 比较前先清理堆顶保证看到的是有效元素 prune(small, True) prune(large, False) # 旧元素逻辑删除 if not small or remove_val -small[0]: balance - 1 else: balance 1 delayed[remove_val] 1 # 新元素插入 if not small or add_val -small[0]: heapq.heappush(small, -add_val) balance 1 else: heapq.heappush(large, add_val) balance - 1 make_balance() ans.append(get_median()) return ans核心步骤我在前面都拆过这里只强调几个容易出问题的地方prune一定要在“判断元素属于哪一堆”之前调用。否则堆顶是一个已删除的脏元素后续比较全部失真。small在极端情况下可能为空所以判断条件写成not small or ...比较安全。make_balance里移动元素前也要prune不然可能把一个已经标记删除的元素当成活元素搬走。这里的get_median每轮都会做一次堆顶清理。你可能担心这样会不会把复杂度搞高。不会因为每个元素最多被延迟标记一次、被真正弹出一次全部加起来的均摊成本是 (O(\log k))整体 (O(n\log k))。3.5 窗口边界与重复元素最容易翻车的地方重复元素是这题的隐藏难点。如果窗口里有好几个相同的值delayed 计数器就体现了价值。你不能用布尔值标记“这个元素删了”因为堆里可能还有多个相同的值在排队。必须用计数。比如 delayed[3] 2表示有两个 3 被标记删除但还没弹出。当堆顶遇到 3 时每弹一次减 1直到减成 0 才说明这个值的删除标记全部清空。窗口边界也很容易翻车。移动窗口时右指针从 k 开始左指针对应 right - k。如果你在循环里把左右指针搞反轻则结果错重则数组越界。我建议每次写这种题都先画一个 k 3 的小窗口把下标对应关系写在纸上再开始写循环。4. 面试官视角这道高频题到底在考察什么4.1 一道题把排序、堆、滑动窗口全串起来面试官喜欢拿这种题当高频题因为它不像某些偏难怪题那样考察背诵而是考察你能否把多个基础知识点串成一个完整方案。你会不会先想到排序会这是人之常情。然后你能不能意识到排序在动态场景下代价太高这是对复杂度的敏感度。你知不知道堆能高效维护堆顶极值这是数据结构的基本功。你能不能想到用两个堆维护中位数这是对“中位数本质上只关心中间位置”这个性质的理解。最后滑动窗口里的延迟删除考验的是你在限制条件下灵活改造经典算法的能力。这一连串问题下来候选人是什么水平基本就摸清了。4.2 现场如何从“不会做”到“做出来”如果你在面试现场第一次遇到这题不要慌按下面这个顺序走先说暴力思路每次排序或者维护有序数组后二分插入。分析暴力思路的问题排序太慢有序数组插入是 (O(n))数据量大时不行。问自己能不能只维护中间位置于是想到双堆。写出数据流版本的 addNum 和 findMedian。面试官加码到滑动窗口时先想到删除是难点再提出延迟删除。这五步本身就是在向面试官展示你的思考过程。很多时候最终代码是否一次通过不是最重要的重要的是你有没有结构化地推进问题。我甚至建议即使你会做也要在开头把“排序方案”提一句。面试官能看到你从朴素方案出发而不是上来就背模板这会让你的答案显得更可信。4.3 常见错误清单和排查方法我把这些年见过的高频 bug 整理成一张表错误类型表现原因修复堆顶没清理就取中位数答案偶尔正确偶尔错延迟删除的脏元素挡在堆顶取中位数前先 prunebalance 变化量写反窗口滑动后中位数不对旧元素删除后新元素插入的 balance 逻辑混了统一走“旧元素删除、新元素插入”两步最大堆忘记存负值插入后数据全乱Python 没有内置最大堆所有入 small 的值都取相反数奇数个数时返回两个堆顶平均值边界用例失败没约定“左堆多一个”奇数直接返回左堆堆顶重复元素被提前弹出堆里还有该值却弹少了只记录“是否删除”而不是次数用字典计数如果你上线调不出来最简单的排查方法是用小数据手动模拟。我用得最多的是这几个用例窗口 k 1输入 [1,2,3]结果应该是 [1,2,3]。如果不对说明删除和插入的边界乱了。窗口 k 2输入 [1,2,3,4]结果应该是 [1.5,2.5,3.5]。如果不对说明偶数取平均那边有问题。窗口 k 3输入 [1,3,-1,-3,5,3,6,7]这是 LeetCode 官方示例直接对照答案。4.4 追问变体数据量超大时的中位数这是对顶堆场景最常见的追问。如果数据流非常大甚至多个机器分布存储你没法把所有数据丢进两个堆怎么办常见的思路是引入分桶统计维护值的频次分布或者维护分位数近似。面试如果走到这里重点不是让你实现一个完美的外部排序方案而是考察你知不知道对顶堆的边界它的空间是 (O(n))必须存全量数据。一旦空间不够就要用近似结构换精度。你回答时可以说“如果单机内存能放得下对顶堆是最直接的选择如果放不下我会考虑分桶统计或者近似分位数但精度会受影响。”这句话就能体现边界意识。5. 我刷这道题到今天的一些实操心得5.1 别背模板背两个不变量我一开始也走过背模板的弯路但这题最大的问题是模板稍有改动就会崩。后来我发现真正要记住的不是代码而是两个不变量左堆所有元素小于等于右堆所有元素。左右堆大小相差不超过 1且中位数位置固定在左堆堆顶或两堆顶平均。所有代码操作都是在维护这两个不变量。你每次写完一个函数先停下来问自己这两个条件现在还成立吗成立代码大概率没错不成立不需要看具体实现就知道 bug 在哪。刷题的时候把这个思维练成条件反射比背十遍代码都管用。5.2 调试这类题最实用的小白鼠用例除了 LeetCode 官方示例我还会用一些极端用例来测自己的实现所有数字都相同比如 [5,5,5,5,5]。这能检验延迟删除的计数逻辑因为重复元素在窗口里频繁进出。窗口大小等于数组长度比如 nums [1,2,3], k 3。这要求初始化窗口的逻辑和后续滑动逻辑结果一致。递增序列和递减序列各跑一遍比如 [1,2,3,4,5] 和 [5,4,3,2,1]。这能暴露插入时比较方向写反的问题。负数混合正数比如 [-1,-2,3,4]。很多人最大堆存负值后符号一多就晕这个用例很有必要。我每次写完这类题都会把上面几个用例跑一遍基本能把常见的错都堵住。5.3 写在最后的一点个人建议如果让我用一句话总结这次的分享我会说对顶堆不是一个需要死记硬背的数据结构而是一种“把中位数问题拆成两个极值问题”的思维模型。左堆维护极大值右堆维护极小值中间那条线就是中位数的位置。数据流版本让你掌握这个模型本身滑动窗口版本让你掌握如何在动态删除的约束下改造模型。这两题连起来刷价值远大于单刷两道题。最后分享一个小习惯每道用到堆的题我都会坚持把“堆顶清理”“平衡条件”“复杂度摊还”这三件事单独写进题解笔记。等到面试前几天不需要重新看整段代码只需要看这几个关键词就能快速把思路拉起来。希望这篇拆解也能成为你的题解笔记之一。