295. 数据流的中位数 (Hard)

专题归类： 06-栈与堆 · 12-技巧专题 LeetCode 链接： https://leetcode.cn/problems/find-median-from-data-stream/

---

在线做题： 写 Python 代码并运行 · Java 跳转 LeetCode

题目描述

中位数 是有序整数列表中的中间值。如果列表的大小是偶数，则没有中间值，中位数是两个中间值的平均值。

• 例如 [2,3,4] 的中位数是 3
• 例如 [2,3] 的中位数是 (2 + 3) / 2 = 2.5

设计一个支持以下两种操作的数据结构：

• void addNum(int num) — 从数据流中添加一个整数到数据结构中。
• double findMedian() — 返回目前所有元素的中位数。

示例：

输入：
["MedianFinder","addNum","addNum","findMedian","addNum","findMedian"]
[[],[1],[2],[],[3],[]]
输出：
[null,null,null,1.5,null,2.0]

解释：
MedianFinder mf = new MedianFinder();
mf.addNum(1);
mf.addNum(2);
mf.findMedian();    // 返回 1.5
mf.addNum(3);
mf.findMedian();    // 返回 2.0

---

题目详细分析

• 数据范围： -10^5 ≤ num ≤ 10^5，最多调用 5×10^4 次 addNum 和 findMedian。
• 输入输出特征： 数据流动态插入，每次插入后可能查询中位数。中位数可能是整数也可能带 .5。
• 边界条件：
  • 只有一个数字 → 中位数就是它自己。
  • 只有两个数字 → 中位数是两数平均值/2。
  • 所有数字相同 → 中位数就是该值。
• 核心约束： 动态数据流，不能每次 findMedian 时重新排序（O(n log n) 太慢）。
• 隐藏条件：
  • 需要在 O(log n) 插入 + O(1) 查询的时间复杂度内完成。
  • 双堆技巧是中位数维护的经典解法：将数据分为较大的一半和较小的一半，分别用不同性质的堆维护。
  • 中位数只取决于数据流中间位置的一到两个元素，不需要保留全排序信息。

---

小白版直白理解

就像在一场比赛中实时统计选手得分的中位数。

想象你有两个箱子，所有选手的得分牌不停地递过来：

• 左箱子（大根堆）： 专门放较小的那一半分数。箱子的顶部（最大的）就是这一半的最高分。
• 右箱子（小根堆）： 专门放较大的那一半分数。箱子的顶部（最小的）就是这一半的最低分。

核心规则：左箱子的所有分数都 ≤ 右箱子的所有分数，并且两个箱子的数量差不超过 1。

每当新分数来了：

1. 先放进左箱子
2. 把左箱子顶部（最大）移到右箱子（保证左箱所有数 ≤ 右箱所有数）
3. 如果右箱子数量比左箱子多，再把右箱子顶部（最小）移回左箱子（平衡数量）

这样：

• 如果总数量是奇数，左箱子顶部就是中位数
• 如果总数量是偶数，左箱子顶部 + 右箱子顶部 的平均数就是中位数

---

解题思路

思路一：双堆法（推荐）

核心思想： 用两个堆：

• 大根堆 small（存负值模拟）： 存储较小的一半元素，堆顶是这一半的最大值。
• 小根堆 large： 存储较大的一半元素，堆顶是这一半的最小值。

维护两个堆的大小平衡（small 至少和 large 一样多），且 small 中所有元素 ≤ large 所有元素。

from heapq import heappush, heappop
 
class MedianFinder:
    def __init__(self):
        self.small = []   # 大根堆（存负值），存较小的一半
        self.large = []   # 小根堆，存较大的一半
 
    def addNum(self, num: int) -> None:
        # 第一步：先插入 small
        heappush(self.small, -num)
        # 第二步：将 small 的最大值移到 large（保证 small ≤ large）
        heappush(self.large, -heappop(self.small))
        # 第三步：平衡大小。如果 large 更多，移回 small
        if len(self.large) > len(self.small):
            heappush(self.small, -heappop(self.large))
 
    def findMedian(self) -> float:
        if len(self.small) > len(self.large):
            return -self.small[0]                    # 奇数，small 堆顶
        return (-self.small[0] + self.large[0]) / 2  # 偶数，两堆顶平均

复杂度： addNum O(log n)，findMedian O(1)，空间 O(n)

思路二：双堆法（另一种平衡策略）

先插入 large，再平衡到 small，逻辑对称。

from heapq import heappush, heappop
 
class MedianFinder:
    def __init__(self):
        self.small = []   # 大根堆
        self.large = []   # 小根堆
 
    def addNum(self, num):
        # 先插入 large
        heappush(self.large, num)
        # 将 large 的最小值移到 small
        heappush(self.small, -heappop(self.large))
        # 平衡：保持 len(small) >= len(large)
        if len(self.small) - len(self.large) > 1:
            heappush(self.large, -heappop(self.small))
 
    def findMedian(self):
        if len(self.small) > len(self.large):
            return -self.small[0]
        return (-self.small[0] + self.large[0]) / 2

两种策略本质上相同，只是第一步的插入方向不同。

思路三：插入排序 + 二分查找（对比用）

维护一个有序列表，每次插入时用二分查找找到插入位置，然后插入。插入操作为 O(n)。

import bisect
 
class MedianFinder:
    def __init__(self):
        self.nums = []
 
    def addNum(self, num):
        bisect.insort(self.nums, num)   # O(n) 插入
 
    def findMedian(self):
        n = len(self.nums)
        mid = n // 2
        if n % 2 == 1:
            return self.nums[mid]
        return (self.nums[mid - 1] + self.nums[mid]) / 2

复杂度： addNum O(n)，findMedian O(1)。插入 O(n) 在数据量大时效率低，但实现简单。

---

易错点

• 大根堆的模拟： Python 的 heapq 只提供小根堆，大根堆需要存负值来模拟。注意在使用 small[0] 时，取出的值需要取负才是真正的最大值。
• 堆的平衡条件： 必须保证 len(small) >= len(large) 且 len(small) - len(large) <= 1。如果平衡条件不同，中位数的计算方式也要相应调整。
• 三步插入法的顺序： 先入 small → 移最大值到 large → 平衡 这三步的顺序不能错。如果跳步，可能导致 small 中的元素大于 large 中的元素，违反核心约束。
• 数据类型： 当总数为偶数时，中位数可能是 .5，需要用 float 或 / 2.0 来计算。Python 中 / 默认返回 float。
• 空堆处理： 题目保证不会在空数据结构中查询中位数，但如果在工程中实现需要处理空值。

---

框架提炼

双堆维护中位数模板：

from heapq import heappush, heappop
 
class MedianFinder:
    def __init__(self):
        self.left = []   # 大根堆（存负值），存放较小的一半
        self.right = []  # 小根堆，存放较大的一半
 
    def addNum(self, num):
        """向数据流中添加一个数"""
        # 1. 插入左堆
        heappush(self.left, -num)
        # 2. 维持 left ≤ right 的特性
        heappush(self.right, -heappop(self.left))
        # 3. 平衡两堆大小
        if len(self.right) > len(self.left):
            heappush(self.left, -heappop(self.right))
 
    def findMedian(self):
        """返回当前中位数"""
        if len(self.left) > len(self.right):
            return -self.left[0]
        return (-self.left[0] + self.right[0]) / 2.0

双堆技巧的通用场景：

• 动态数据流中的中位数（本题）
• 数据流中的第 k 大元素（一个大小为 k 的最小堆）
• 滑动窗口中的中位数（双堆 + 延迟删除）

---

关联题目

• 215-数组中的第K个最大元素 — 与 295 的区别：215 是静态数组，一次查找第 k 大；295 是动态数据流，持续查找中位数（相当于动态第 n/2 大）。
• 347-前K个高频元素 — 同样是堆的应用，但 347 用哈希表 + 最小堆，295 用双堆。两者展示了堆在不同场景下的灵活运用。
• 480-滑动窗口中位数 — 295 的升级版：在滑动窗口（固定大小）中维护中位数。需要双堆 + 延迟删除，实现复杂度大幅提升。