算法面试：如何求数据流的动态中位数？

题干与适用场景

设计 MedianFinder，支持两个操作：

• addNum(num)：向整数数据流加入一个数。
• findMedian()：返回目前所有数字的中位数；奇数个元素时返回中间值，偶数个元素时返回两个中间值的平均数。

假设只有插入，没有删除；findMedian() 只会在至少插入一个数字后调用。输入可以包含负数、重复值和 32 位有符号整数，最多执行 50,000 次操作。目标是让每次插入为 O(log n)，查询为 O(1)，空间为 O(n)。

例如，依次插入 5、2、10、4，每次插入后的中位数分别是 5、3.5、5、4.5。如果每次查询前重新排序， 结果正确，但查询成本是 O(n log n)；如果始终维护有序数组，查询只要 O(1)，中间插入却需要移动 O(n) 个元素。

2026 年的公开面试准备材料仍把这道题作为两堆模式的代表性编码题。它适合通用软件工程师、后端、数据和 基础设施岗位的算法轮。核心考点不是背出“最大堆加最小堆”，而是能否从查询目标推导数据结构、明确两个 不变量，并解释固定的搬运顺序为什么不会破坏分区。

面试官考察点

第一个信号是候选人能否从操作比例反推结构。中位数只与排序后的中间位置有关，不需要维护完整顺序。为了 O(1) 查询，两个中间候选必须始终位于可直接读取的位置；堆顶正好提供这种能力。

第二个信号是能否同时维护“分区”和“平衡”：

1. 较小的一半放在最大堆 lower，较大的一半放在最小堆 upper，且 lower 中每个值都不大于

upper 中每个值。

1. lower 的元素数等于 upper，或恰好多一个。

这两个条件缺一不可。只保证数量接近，两个堆可能装反；只保证左右顺序，堆大小可能相差很多，堆顶就不再是 中间位置。

第三个信号是复杂度表达是否准确。一次插入做常数次堆操作，每次 O(log n)；查询只读一个或两个堆顶， 所以是 O(1)。保存全部输入仍需要 O(n) 空间。这里的“流式”表示在线更新，不代表内存是常数。

最后看验证能力。强回答会主动测试首个元素、偶数和奇数长度、重复值、全负数、单调递增、单调递减和整数 极值，还会用一个慢但明显正确的有序数组实现做随机差分，而不是只跑题目示例。

回答前需要澄清的问题

• 只有插入，还是还要删除旧元素？ 只有插入时两个普通堆足够；滑动窗口需要延迟删除或支持有序多重集合。
• 查询会在空数据流上调用吗？ 本题保证不会。若生产接口允许空流，返回可选值或抛出明确异常，不能读取空堆顶。
• 输入是整数还是浮点数？ 本题使用整数。浮点数若允许 NaN，比较关系不再是全序，必须先定义拒绝或排序规则。
• 偶数个元素如何定义中位数？ 本题取两个中间值的算术平均数，因此返回类型应支持小数。
• 需要精确中位数还是允许近似？ 本题要求精确值，所以必须保留足够的信息；无界流且内存受限时需要改用分位数摘要。
• 数值范围是否会让平均值溢出？ Python 整数不会溢出；固定宽度语言应先提升到更宽类型，再做加法和除法。
• 查询和插入的比例如何？ 高频查询适合两堆；如果只在所有输入结束后查询一次，收集后排序通常更简单。
• 是否需要并发安全？ 当前实现是单线程。并发版必须让两个堆的一次搬运和一次查询看到同一状态快照。

30 秒回答框架

“我把较小的一半放进最大堆 lower，较大的一半放进最小堆 upper。始终保持 lower 的所有值不大于 upper，并让 lower 的大小等于 upper 或多一。插入时先放进 lower，再把 lower 最大值移到 upper，这样修复左右顺序；如果 upper 变大，再把它的最小值移回 lower，恢复大小不变量。奇数个 元素时中位数是 lower 堆顶，偶数个元素时是两个堆顶的平均数。插入做常数次堆操作，所以是 O(log n)；查询 O(1)，空间 O(n)。”

分步骤深入解答

第一步：比较基础方案，确认瓶颈在哪里。

| 方案 | 插入 | 查询中位数 | 空间 | 适用条件 | |---|---:|---:|---:|---| | 无序数组，查询时排序 | O(1) | O(n log n) | O(n) | 几乎不查询，只在末尾求一次 | | 始终维护有序数组 | O(n) | O(1) | O(n) | 数据量小，代码简洁比插入性能重要 | | 带顺序统计的平衡树 | O(log n) | O(log n) 或更好 | O(n) | 还需要删除、排名或任意分位数 | | 最大堆加最小堆 | O(log n) | O(1) | O(n) | 只插入，持续查询精确中位数 |

二分查找能在 O(log n) 找到有序数组的插入位置，却不能消除数组移动成本。普通平衡树能保持有序，但如果 节点没有子树大小，就不能直接定位第 k 个元素。两堆只维护中位数需要的两个边界，因此是当前契约下最小 且完整的结构。

第二步：把中位数改写成两个堆顶。

令 lower 保存排序后较小的一半，用最大堆，使其中最大值随时位于堆顶；upper 保存较大的一半，用 最小堆，使其中最小值位于堆顶。规定 lower 可以比 upper 多一个元素：

元素总数为奇数：lower 比 upper 多 1，中位数 = max(lower)
元素总数为偶数：两个堆一样大，中位数 = (max(lower) + min(upper)) / 2

Python 的通用 heapq 接口以最小堆为基础。为了兼容常见 Python 版本，代码把 lower 的值取负数保存： 逻辑最大值 x 对应数组中的最小负数 -x，所以 -lower[0] 是较小一半的最大值。

第三步：使用固定的“放入、跨堆、再平衡”顺序。

与其为“新值属于左边还是右边”写多组分支，可以固定执行：

1. 把 num 取负后压入 lower。
2. 弹出 lower 的逻辑最大值，放入 upper。
3. 如果 upper 比 lower 大，把 upper 的最小值移回 lower。

import heapq


class MedianFinder:
    def __init__(self) -> None:
        self.lower = []  # 负数模拟最大堆，保存较小的一半
        self.upper = []  # 最小堆，保存较大的一半

    def add_num(self, num: int) -> None:
        heapq.heappush(self.lower, -num)

        largest_lower = -heapq.heappop(self.lower)
        heapq.heappush(self.upper, largest_lower)

        if len(self.upper) > len(self.lower):
            smallest_upper = heapq.heappop(self.upper)
            heapq.heappush(self.lower, -smallest_upper)

    def find_median(self) -> float:
        if not self.lower:
            raise ValueError("median is undefined for an empty stream")

        if len(self.lower) > len(self.upper):
            return float(-self.lower[0])

        return (-self.lower[0] + self.upper[0]) / 2.0

这个顺序看似多搬一次，却减少了容易写错的条件组合。实现也可以先比较 num 与 -lower[0]，再选择目标 堆并按大小搬运；两种写法复杂度相同。面试中应选择自己更容易证明和复查的版本。

第四步：证明顺序不变量。

假设插入前，每个 lower 元素都不大于每个 upper 元素。把新值暂时放进 lower 后，只有新值可能放错 半区。此时弹出 lower 的最大值：

• 留在 lower 的所有值都不大于被弹出的值。
• 旧 lower 的值原本都不大于旧 upper。
• 因此把这个最大值加入 upper 后，新的 lower 仍不大于新的 upper 中任何值。

第二步后 upper 可能比 lower 多一个。若发生这种情况，移回 upper 的最小值。它不大于 upper 剩余的每个值，又不小于原 lower 的最大边界，所以左右顺序仍成立。最终两堆大小相等，或 lower 多一个，平衡不变量恢复。

初始化时两个堆为空，不变量成立；每次插入都保持不变量，因此由归纳法可知，任意操作序列后堆顶都对应排序 结果的中间位置。

第五步：走一遍包含跨区移动的例子。

插入 5： lower = [5]       upper = []        median = 5
插入 2： lower = [2]       upper = [5]       median = 3.5
插入 10：lower = [5, 2]    upper = [10]      median = 5
插入 4： lower = [4, 2]    upper = [5, 10]   median = 4.5

堆内部不保证完整排序，[4, 2] 只表示堆顶是 4。调试时不能把堆数组打印结果误当成有序数组；需要检查的是 堆性质、两个堆顶和跨堆不变量。

第六步：计算复杂度，并说明简单方案何时更好。

add_num 最多执行五次压入或弹出，每次堆操作为 O(log n)，常数次相加仍是 O(log n)。 find_median 读取长度和堆顶，是 O(1)。每个输入值恰好保存在一个堆中，空间为 O(n)。

如果产品只收集一批数据并在结束时求一次中位数，保存数组后排序的实现更短，也可能有更好的连续内存性能； 没有必要为了一个查询维护在线结构。如果值域固定为 0 到 100，长度 101 的计数数组还能把插入降为 O(1)，查询扫描固定 101 个桶，在该固定值域下也可视为 O(1)。

第七步：用确定用例和随机差分闭环。

固定用例至少包括：

| 输入序列 | 最终中位数 | 主要风险 | |---|---:|---| | [7] | 7 | 首个元素 | | [1, 2] | 1.5 | 偶数平均 | | [2, 2, 2] | 2 | 重复值 | | [-5, -1, -3] | -3 | 负数与取反最大堆 | | [1, 2, 3, 4, 5] | 3 | 单调递增 | | [5, 4, 3, 2, 1] | 3 | 单调递减 | | [-2147483648, 2147483647] | -0.5 | 平均值与整数提升 |

随机测试每次生成一个整数并同时加入 MedianFinder 和参考数组。参考数组排序后直接取中间值；每次插入都 比较两个结果，并断言 len(lower) 只可能等于 len(upper) 或多一。慢实现不适合生产复杂度，却是很好的 正确性 oracle。

高质量示范回答

“我先确认这是只插入的精确中位数，而且查询不会发生在空流上。若还要删除窗口外元素，普通堆不能直接删除 任意值，解法需要改变。

为了让查询是常数时间，我希望排序后中间位置的值一直暴露在数据结构顶端。我会用两个堆：最大堆 lower 保存较小的一半，最小堆 upper 保存较大的一半。维持两个条件：lower 中所有值都不大于 upper，并且 lower 的大小等于 upper 或多一。

插入时我采用固定三步。先把新值放进 lower；再把 lower 最大值移到 upper，保证左右分区正确； 如果 upper 变得更大，就把它的最小值移回 lower。这样每次结束后两个不变量都恢复。奇数长度时 lower 多一个，中位数就是它的堆顶；偶数长度时取两个堆顶的平均数。

每次插入只有常数次堆操作，所以是 O(log n)；查询只读堆顶，是 O(1)；所有数字仍要保留，空间是 O(n)。我会测试单元素、偶数、重复值、负数、单调序列和整数极值，再与每次排序的慢实现做随机差分。 如果所有数据都在 0 到 100，我会改用 101 个计数桶；如果只在最后查询一次，直接排序会更简单。”

常见错误

• 只让两个堆大小接近 → 左右值可能交叉，堆顶不是两个中间值 → 同时维护跨堆顺序和大小不变量。
• 把较小一半放进最小堆 → 堆顶暴露的是全局最小值，不是左半区最大值 → 较小一半使用最大堆。
• 偶数长度只返回一个堆顶 → 中位数定义错误 → 大小相等时取两个堆顶的平均数。
• 平均前使用固定宽度整数相加 → 两个大整数可能先溢出 → 先提升到更宽或浮点类型再相加。
• 认为二分插入有序数组是 O(log n) → 找位置虽快，移动元素仍是 O(n) → 把查找和写入成本分开计算。
• 把 Python 堆的底层数组当成完整有序序列 → 测试会得出错误结论 → 只依赖堆顶和父子堆性质。
• 对空堆直接读取索引 0 → 运行时异常位置不清晰 → 明确禁止空查询，或在接口边界返回可选值。
• 声称在线算法只用常数空间 → 两个堆仍保存全部输入 → 精确中位数空间是 O(n)。
• 滑动窗口仍使用相同代码 → 过期元素可能留在堆顶并污染结果 → 加入延迟删除和有效大小，或改用有序多重集合。
• 只测试一个示例 → 取反、重复值和再平衡错误可能未触发 → 固定边界用例加随机差分。

追问及应对

追问一：如果所有整数都在 0 到 100 之间，如何优化？

维护长度 101 的计数数组和总元素数。插入只增加一个桶，时间 O(1)；查询根据两个中间排名从小到大扫描 101 个桶，固定值域下时间和空间都可视为 O(1)。如果值域上限会随输入增长，扫描成本应写成 O(R)， 其中 R 是值域大小，不能继续称为常数。

追问二：如果 99% 的值在 0 到 100，剩余值可能任意大呢？

可以为区间内保留 101 个计数桶，并分别用支持排名选择的有序结构保存小于 0 和大于 100 的异常值。查询时先 根据异常值数量判断目标排名落在左异常区、固定区间还是右异常区，再在对应结构中选择。只有普通堆却没有 排名能力时，不能因为“99% 在范围内”就宣称查询为常数时间；极端情况下中位排名仍可能落入异常值结构。

追问三：如何求最近 k 个数字的滑动窗口中位数？

窗口滑动时要删除离开的值。二叉堆不能高效定位任意元素，常见做法是两个堆加延迟删除表：逻辑删除时记录 计数并更新两个堆的有效大小，只有待删除值到达堆顶时才真正弹出；每次读取中位数前清理两个堆顶。更新为 摊销 O(log k)，查询堆顶仍为 O(1)。支持重复值和删除的平衡多重集合实现更直接，但依赖语言库。

追问四：无界数据流且内存固定，还能返回精确中位数吗？

对任意整数流，一般不能在固定内存中始终保留精确中位数，因为被丢弃的历史值可能在未来决定中间排名。需要 把契约改为近似分位数，使用有误差界的分位数摘要，并明确允许的排名误差、置信要求和合并方式。此时答案不再 是精确的两个堆。

追问五：如何支持多个线程同时插入和查询？

两个堆构成一个逻辑状态。最简单的正确方案是在完整的 addnum 和 findmedian 外使用同一把互斥锁， 避免查询看到“已经从 lower 弹出、尚未放入 upper”的中间状态。若读流量很高，可以发布不可变的中位数 快照，但快照更新频率会引入新鲜度取舍，必须写进接口契约。

追问六：多个分片各自给出中位数，能否直接合并成全局中位数？

不能。分片中位数丢失了分片大小和分布信息；即使取加权平均，也不等于全局中位数。精确结果需要能回答全局 排名的结构，例如按值域汇总计数并做分布式选择；近似结果则使用可合并的分位数摘要。先确定精度和延迟目标， 再选择全局结构。