刚刷到LeetCode热题100里这个“贪心算法篇”的时候我其实是有点疑惑的。数组中的第K个最大元素、前K个高频元素、数据流的中位数——这三道题从表面上看和课本里经典的贪心场景区间调度、哈夫曼编码、找零钱完全不是一回事。第K大元素怎么个“贪心”法数据流中位数又是在贪什么把三道题都刷完、又对比了各路题解之后我才逐渐反应过来这个分类其实不是严格意义上的算法分类而是想让你形成一种“用最小的代价维护局部最值”的思维习惯。贪婪地只保留自己关心的那部分信息丢掉其余的大部分数据这在工程上的收益往往比所谓的“标准答案”更重要。这篇文章就把这三道题放在一起拆一遍。我会先讲清楚为什么它们会被归进贪心篇再逐题给出从暴力到最优的完整推导最后把三题串联成一个“动态TopK”的通用模型。不管你是准备面试还是想提升工程里的数据处理能力这套内容都值得花半小时消化。1. 被标题“骗”了这三道题真正的底层是堆的局部最优思维1.1 为什么LeetCode把这三题放进贪心算法篇先回答最直接的困惑。LeetCode的题目分类经常会按“推荐学习路径”而不是严格的数据结构标签来划分。热题100的贪心篇里塞进这三道题真正想让你练的其实是同一个核心能力在数据源源不断出现的时候如何只做局部最优决策就能维护出全局想要的信息。用第K个最大元素举例。如果用一个大小为K的小顶堆来维护“当前已经见过的元素中最大的那K个”每来一个新元素你只需要做一次比较如果新元素比堆顶也就是当前第K大还小直接丢掉如果比堆顶大就把堆顶替换掉。这个过程里你每一步都只是在做“当前最值层面的小决策”但最终却精确拿到了全局第K大。从“决策方式”上看这确实带着贪心味道——每一次操作只关心当前这个元素值不值得进入我维护的“候选集”从不回退从不重新审视已经丢掉的元素。这就是“只看当下最优”的典型思路。数据流的中位数也是同理。维护一个大顶堆存较小的一半、一个小顶堆存较大的一半每次来一个数判断它该进哪边、需不需要调整平衡——每一步都是局部判断但堆顶的组合随时能算出全局中位数。所以别纠结“这是不是严格意义上的贪心”。这三道题是“贪心思想 堆结构”的完美教学案例重点在于理解“如何用局部决策代替全量排序”。这个思维在真实的后端系统里非常值钱比如实时榜单、热门文章TopK、日志频率统计全是这套东西。1.2 堆和优先队列所有解法共用的一套地基既然三道题都绕不开堆先把堆这个东西说透。堆在物理上就是一个数组但逻辑上是一棵完全二叉树。最大堆保证父节点不小于任意子节点最小堆保证父节点不大于任意子节点。这意味着堆的根节点永远是全局最大或最小值而且插入一个元素、弹出堆顶元素都只需要O(logN)的时间。工程里我们一般不用手写堆直接用语言自带的优先队列语言默认优先队列自定义为最大堆Pythonheapq.heappush / heapq.heappop默认最小堆存负数或者自定义对象比较JavaPriorityQueue默认最小堆new PriorityQueue(Comparator.reverseOrder())Cpriority_queue默认最大堆priority_queueint, vector , greater 有一个非常容易踩的坑Python的heapq不提供直接的最大堆接口很多人一上来就直接heappush结果发现取出来的永远是最小值。我的习惯是遇到需要最大堆的场景就统一存负值比如要维护最大的K个值就heappush(heap, -num)取出来的时候再取负。这个操作看起来土但工程里最不容易出错。手写堆在面试里也可能会被问到但更常考的是“你知不知道PriorityQueue底层怎么实现的”“插入和弹出的复杂度分别是多少”。能答出“上浮”“下沉”“完全二叉树数组存储”基本就是过关了。三道题里第K大元素和TopK高频元素本质上是“静态数组上的多次局部最优”数据流中位数则是“动态数据流上的持续局部最优”。工具和思维模型统一了解题就差临门一脚了。2. 数组中的第K个最大元素从全量排序到快速选择的降维2. 数组中的第K个最大元素三种方案的复杂度博弈2.1 傻瓜方案与堆方案从O(NlogN)到O(NlogK)先看题目本身给定一个无序数组nums返回数组中第K个最大的元素。比如[3,2,1,5,6,4]K2答案就是5。最容易想到的做法是排序def findKthLargest(nums, k): nums.sort(reverseTrue) return nums[k - 1]这行代码能跑但有两个问题。第一它把数组全部排好序了而你可能只需要确认一个位置的元素第二排序的时间是O(NlogN)如果数组特别大、K又很小这个复杂度没必要。比如你有1000万条日志想找第10大的错误码全量排序就是浪费。堆解法就能把复杂度降下来import heapq def findKthLargest(nums, k): heap [] for num in nums: heapq.heappush(heap, num) if len(heap) k: heapq.heappop(heap) return heap[0]这段代码维护了一个大小为K的小顶堆。小顶堆的堆顶永远是堆里最小的那个元素。当堆里塞满K个元素后每来一个新元素如果它比堆顶大就替换进去如果比堆顶小它连进入候选集的资格都没有。遍历完整个数组堆顶就是“前K大元素里的最小值”也就是第K大的元素。时间复杂度是O(NlogK)。当K远小于N时这个优势非常明显。如果KN那堆解法退化成O(NlogN)和排序没区别这种时候就别硬用堆了。想验证自己理解没理解可以看个例子数组[3,2,1,5,6,4]K2。过程是3入堆2入堆1来了比堆顶2小丢弃5来了比堆顶2大替换掉26来了比堆顶3大替换4来了比堆顶5小丢弃。最后堆里是[5,6]堆顶就是5。每一步都只做局部判断但结果全局正确。2.2 快速选择平均O(N)的实现与退化风险堆解法的O(NlogK)已经不错了但面试官通常会追问一句能不能做到O(N)答案是能用快速选择QuickSelect。思路脱胎于快速排序的partition操作。随便选一个pivot把数组分成“比pivot大的”和“比pivot小的”两拨。如果pivot最终落在第K个最大元素的坑位上直接返回如果pivot的位置比目标靠前说明第K大的元素在右边那一拨反之在左边那一拨。这样每次只需要递归处理一侧平均复杂度是N N/2 N/4 ... O(N)。import random def findKthLargest(nums, k): def quick_select(left, right): pivot_idx random.randint(left, right) nums[pivot_idx], nums[right] nums[right], nums[pivot_idx] pivot nums[right] # partition把大于pivot的放左边小于的放右边 # 结束后i指向pivot在新数组中的下标 i left for j in range(left, right): if nums[j] pivot: nums[i], nums[j] nums[j], nums[i] i 1 nums[i], nums[right] nums[right], nums[i] if i k - 1: return nums[i] elif i k - 1: return quick_select(i 1, right) else: return quick_select(left, i - 1) return quick_select(0, len(nums) - 1)要注意的是快速选择的最坏情况是O(N^2)发生在每次选的pivot都恰好是最小值或最大值的时候。这也是为什么代码里要加一行random.randint随机化pivot能把最坏情况出现的概率降到极低。面试里如果你只写固定取第一个元素当pivot面试官大概率会追问“如果输入是已经排好序的数组会怎样”。这时候能答出随机化和三数取中是加分项。我在牛客和LeetCode评论区都见过有人争论快速选择和堆哪个更好。我个人的判断标准是如果题目允许修改原数组且数据量特别大、对时间极度敏感快速选择是首选如果数据是流式到达的或者题目明确不能修改数组或者要的不是一次性的第K大而是“不断查询”那必须用堆。这两种思路不冲突它们是应对不同场景的工具不是非此即彼。2.3 一个容易翻车的细节第K大和第N-K1小刷这道题的时候不少人会在下标处翻车。第1大的元素在从大到小排序后是第0个下标第K大的元素是第K-1个下标。但如果换一种写法把数组按从小到大排序那第K大对应的就是第len(nums) - k个下标。两种写法都对但混着写就很容易数组越界。我强烈建议写代码前在注释里先写清楚“我这里用的是从大到小排序的哪个下标”再动手写partition。不要一边写一边换算做算法题最怕的就是这种低级错误。还有一点LeetCode这道题的原题有一个附加条件K是有效的即1 K nums.length。但实际面试中可能会出现K非法的情况比如K0或者K比数组长度还大。写代码时加上一个健壮性判断永远是加分项哪怕只是简单的if k 1 or k len(nums): return -1。3. 前K个高频元素哈希表与堆的组合拳3.1 构建频率表这一步很多教程一笔带过题目是给定一个非空整数数组返回其中出现频率前K高的元素。比如[1,1,1,2,2,3]K2返回[1,2]。第一步基本是固定套路用一个哈希表统计每个数字出现的次数时间复杂度O(N)from collections import Counter freq Counter(nums)在Python里直接用Counter一行搞定。如果是Java就是HashMapInteger, Integer遍历数组map.put(num, map.getOrDefault(num, 0) 1)。但很多人做到这一步就开始想“怎么对频率排序”了而且第一反应是“把频率放进一个数组里然后对整个数组排序”。这个思路能做但不是最优。因为HashMap里可能有几十万个键值对而我们只需要前K个。对所有键值对排序的时间复杂度是O(MlogM)M是不同元素的个数在M远大于K时是很不划算的。我们真正需要的是“只维护前K大”这又回到了第一题里的局部最优思维。3.2 小顶堆维护TopK为什么“小”顶堆反而能找“大”元素这一步是整道题最反直觉的地方我们想要的是“频率最高的K个元素”听起来应该用大顶堆把频率最大的放在堆顶然后取K次。但如果你真这么做了复杂度会是O(MlogM)——因为你得先把所有元素都塞进大顶堆再连续弹出K次。当K接近M时这没什么问题但K远小于M时你用大顶堆存了太多根本不关心的数据。正确做法是用小顶堆让堆的大小始终不超过Kimport heapq from collections import Counter def topKFrequent(nums, k): freq Counter(nums) heap [] for val, cnt in freq.items(): heapq.heappush(heap, (cnt, val)) if len(heap) k: heapq.heappop(heap) return [val for _, val in heap]这里堆的元素是(频率, 元素值)这样的元组。Python在比较元组时会先比较第一个元素也就是频率如果频率相同再比较第二个元素。堆顶永远是“前K个高频候选中频率最低的那个”。如果遇到一个频率更高的新元素它就会把堆顶挤出去。最后堆里剩下的就是频率最高的K个元素。这里有个口味问题如果两个元素频率相同返回哪个LeetCode原题是不要求顺序的但实际面试时面试官可能会追问“如果要求频率相同的按元素值从小到大输出怎么办”。这时候可以提前排序或者自定义比较器但最稳妥的回答是先确认题目是否对顺序有要求不要自作主张。我见过候选人因为这个细节被追问了三轮。Java版本的堆写法也比较容易踩坑。PriorityQueue默认小顶堆但如果你直接塞Map.Entry进去需要自己写比较器PriorityQueueInteger heap new PriorityQueue((a, b) - freq.get(a) - freq.get(b)); for (Integer key : freq.keySet()) { heap.add(key); if (heap.size() k) { heap.poll(); } }注意freq.get(a) - freq.get(b)这个写法是有溢出风险的虽然在这个题目的数据范围内不会出问题但严谨的做法是用Integer.compare(freq.get(a), freq.get(b))。面试里写出这个细节至少能留下一个“这家伙写代码很稳”的印象。3.3 另一种O(N)路径桶排序解法堆解法的时间复杂度是O(NlogK)已经是面试标准答案了。但还有个更极致的做法桶排序思想。频率最多不可能超过N数组长度所以可以建一个长度为N1的桶数组下标代表频率桶里放对应频率的元素。然后从后往前遍历桶先碰到的高频元素就是答案。def topKFrequent_bucket(nums, k): freq Counter(nums) buckets [[] for _ in range(len(nums) 1)] for val, cnt in freq.items(): buckets[cnt].append(val) res [] for i in range(len(nums), 0, -1): for val in buckets[i]: res.append(val) if len(res) k: return res桶排序的时间是O(N)比堆解法更优但代价是额外开了一个长度N1的数组。当数组特别大时这个空间开销可能不可接受。而且如果大多数元素频率都集中在一两个桶里其他桶全是空的这个方案的空间利用率也很低。所以面试时如果你先写了堆解法再补充一句“如果空间允许还能用桶排序优化到O(N)”面试官通常会觉得你思路很开阔。但反过来如果一上来就写桶排序很可能被追问“空间复杂度是多少”“最坏情况下桶的规模是多大”答不好反而扣分。我个人建议是堆解法作为默认答案桶排序作为拓展思路。两个都准备看情况出招。4. 数据流的中位数双堆平衡动态数据的定海神针4.1 为什么朴素方案扛不住“流式”数据这道题的描述很有意思设计一个支持以下两种操作的数据结构——addNum(int num)向数据结构中添加一个整数findMedian()返回目前所有元素的中位数。数据流意味着每时每刻都会有新元素进来要求你能随时回答“当前中位数是多少”。最容易想到的方案是维护一个动态数组每次新元素加入后做插入排序或者全排序然后取中间位置的值。这在小数据量时没有任何问题但如果你看了这两个操作的时间复杂度就会发现问题很大插入排序O(N)全排序O(NlogN)。当数据流连续不停、每秒进来几千个数的时候这个方案很快就会被拖垮。另一种思路是维护两个堆让“中位数”变成两个堆顶的简单计算。这个思路之所以优雅是因为它把“动态插入 查询”的复杂度压到了O(logN)插入、O(1)查询而且完全不需要像平衡树那样大动干戈。4.2 两个堆的分工一个管左半边一个管右半边理解这个解法只需要一句话把数据分成两半左半边的最大值和右半边的最小值一夹中位数就出来了。具体来说大顶堆small存放所有元素中较小的一半堆顶是这一半的最大值。小顶堆large存放所有元素中较大的一半堆顶是这一半的最小值。如果所有元素个数是奇数那中位数就在元素较多的那个堆的堆顶如果是偶数中位数是两个堆顶的平均值。关键是不管数据流里来了多少个数都要让两个堆满足两个条件small中所有元素都不大于large中所有元素两个堆的元素数量差不超过1。import heapq class MedianFinder: def __init__(self): self.small [] # 大顶堆存负数值 self.large [] # 小顶堆存正常值 def addNum(self, num): # 先按值的大小决定进哪个堆 if not self.small or num -self.small[0]: heapq.heappush(self.small, -num) else: heapq.heappush(self.large, num) # 保证small的最大值不超过large的最小值 if self.small and self.large and -self.small[0] self.large[0]: val -heapq.heappop(self.small) heapq.heappush(self.large, val) # 平衡堆大小 if len(self.small) len(self.large) 1: val -heapq.heappop(self.small) heapq.heappush(self.large, val) if len(self.large) len(self.small): val heapq.heappop(self.large) heapq.heappush(self.small, -val) def findMedian(self): if len(self.small) len(self.large): return -self.small[0] return (-self.small[0] self.large[0]) / 2这段代码有一个细节值得展开讲为什么在判断“该进哪个堆”的时候不是按num和当前堆的大小关系随便插而是要先和small的堆顶比较因为-self.small[0]是较小一半中的最大值。如果新来的数比这个最大值还小那它显然属于左半边直接进small否则它属于右半边进large。这个判断保证了数据始终能按大小切进正确的堆。还有一种更简洁的写法是利用“先插再调”的方式每次新元素先插入large然后把large中最小的那个移到small如果small数量超过large再把small中最大的移到large。这样写代码更短但理解起来没有上面版本直观。我建议初学先按“判断进哪边再平衡”的版本写等彻底理解了再考虑精简。4.3 工程场景延伸与面试追问数据流中位数这套“双堆”模型应用场景远不止LeetCode题目本身。举几个我实际接触过的例子监控系统里要实时看“当前线上99%的请求耗时”不可能把每个请求的耗时都存下来排序但可以用两个堆动态维护头部数据。推荐系统评分卡的实时反馈要随时知道用户群体的中位数评分双堆可以做到即时响应。排行榜系统要维护Top100本质上就是第一题堆解法的工程化。面试里这道题最常见的追问有三个第一“如果中位数相同但是有大量重复元素怎么办”答案是堆里允许重复值不影响正确性只是堆调整时稍微多一点操作而已。第二“如果内存放不下所有历史数据怎么办”这就超出双堆的能力边界了通常的解法是分桶统计类似于把数据按大小范围分桶记录每个桶的计数然后根据桶的累计数量推算中位数所在区间再在桶内细化。这种问题考的是你是否知道“当数据结构装不下时精度换空间”的工程思维。第三“为什么不用有序数组加二分查找”因为数组插入是O(N)的。平衡树可以做到O(logN)插入但实现复杂度比双堆高得多。双堆方案在“只求中位数”这个场景下是代码量和复杂度之间的最优平衡。5. 三道题串成一条线TopK问题的通用模型5.1 复杂度对比一张表看清思路选择逻辑把三道题放在一起看你会发现它们本质上是同一个模型的不同变体题目核心数据结构时间复杂度空间复杂度适用场景数组中的第K个最大元素快速选择 / 大小为K的小顶堆O(N)平均 / O(NlogK)O(1) / O(K)静态数据、一次性查询前K个高频元素哈希表 大小为K的小顶堆O(NlogK)O(N)静态数据、按频率筛选数据流的中位数大顶堆 小顶堆插入O(logN)、查询O(1)O(N)动态数据、持续查询这个表格想表达的核心结论是凡是“只要前K个”“只要某个位置的最值”这类需求都不要急着全量排序。先用堆把规模限制在K是一种工程上极其常用的优化思路。而流式数据场景就必须做到“每一时刻都维护着答案”双堆结构就是为了这个目的设计的。有人可能会问快速选择O(N)不是比堆的O(NlogK)更快吗为什么TopK高频元素的题不用快速选择因为快速选择需要一个可以在O(1)时间内比较大小的数组元素而“频率”在哈希表里你可以把(频率, 元素)当成一个整体放进数组用快速选择但你要么得新建一个大数组要么得对哈希表的键集合做转换空间和时间都不划算。高频元素这道题里哈希统计成本已经付出去了堆只在K个元素上操作整体最平衡。5.2 从刷题到项目堆的实际用武之地我在之前的文章里也提过LeetCode题目练到后来真正有价值的部分不是“记住每道题的解法”而是形成一套“看到问题先建模”的能力。这三道题就是很好的建模素材当你面对“海量日志里找出频率最高的K个错误码”时应该立刻想到哈希表小顶堆。当你面对“游戏服务器要实时统计全服战力前100的玩家”时应该立刻想到维护一个容量100的小顶堆新玩家战力比堆顶高就替换否则忽略。当你面对“电商大屏要实时显示GMV中位数变化”时应该立刻想到双堆模型。这些场景的共同特点是数据量大、单条数据价值密度低、只关心头部或特定位置的信息。如果你真的在业务里写过类似的东西再回头看这三道题会有一种“原来面试题就是工程题”的感觉。我个人在实践里踩过的一个坑是小顶堆固定容量K的写法在Java里如果用PriorityQueue默认的小顶堆容量超了就poll()。但如果你忘了限制容量把几百万个元素全塞进堆里那堆的性能优势就完全消失了变回了一个普通的排序问题。每次add之后要立刻检查size() K这个检查不能等也不能在最后统一清理。这是一个非常细节但影响巨大的实现习惯。结尾之前的一点个人体会这三道题我当年刷的时候也是一路踩坑过来的。第K大元素的下标换算搞混过前K个高频元素一开始真的用大顶堆把全量数据塞进去了数据流中位数里两个堆谁管哪半边也理不清过。但现在回看这些坑恰恰是最好的老师。尤其是我自己在做后端数据统计的时候小顶堆限容这个写法几乎每周都会用到面试里答“怎么求TopK”也早就从背答案变成了讲清楚不同方案的取舍。如果你正在按专题刷LeetCode我的建议是不要只盯着“这道题的AC代码”而是把同类的三五道题横向对比。比如把这三道题和大文件里的TopK问题、直播弹幕的实时热度统计放一起看你会发现背后的工具和思维完全一致。这比单纯刷完100题更有价值。最后再分享一个小技巧刷这种和“堆”有关的题先在草稿纸上画一下堆的插入和弹出过程尤其是数据流中位数那道题画一遍addNum的完整流程比我在这里写一千字都管用。理解了过程代码自然就写得出来。