不能直接用list.sort()动态求中位数,因为每次插入后排序时间复杂度为O(n log n),在高频数据流下性能急剧下降,且sort()破坏原顺序,无法满足实时“边加边查”需求;正确做法是用双堆(大顶堆存较小半、小顶堆存较大半)维护,通过即时插入与跨堆迁移保证平衡,使addNum为O(log n)、findMedian为O(1)。

为什么不能直接用 list.sort() 动态求中位数
每次新来一个数就 list.append() 再 list.sort(),时间复杂度是 O(n log n),在数据流持续到达(比如每秒上千条)时会迅速拖慢整体性能。更关键的是,sort() 破坏了原有顺序,无法支撑“边加边查”的实时性要求。
真正可行的思路是把数据逻辑拆成两半:一半存较小的数(用最大堆),一半存较大的数(用最小堆)。Python 没有原生最大堆,但可以用 heapq 存负值模拟——这是最常被忽略的起点。
- 小顶堆(
large):存较大一半,heapq.heappush(large, x)直接可用 - 大顶堆(
small):存较小一半,存-x,取时再取负:-heapq.heappop(small) - 始终维持
len(large) == len(small)或len(large) == len(small) + 1,中位数就落在堆顶
如何用 heapq 维护两个堆并保证平衡
核心不是“插入完再平衡”,而是在每次 addNum() 时立刻决定该数去哪边、是否需要跨堆迁移。常见错误是先无脑 push 再统一 rebalance,导致堆顶值不反映真实分割点。
推荐做法:先压入 small(作为临时缓冲),再弹出其最大值(即 -heapq.heappop(small))压入 large;接着检查 large 是否过大,若 len(large) > len(small) + 1,就把 large[0] 弹给 small(记得取负)。
立即学习“Python免费学习笔记(深入)”;
def addNum(self, num: int) -> None:
heapq.heappush(self.small, -num)
heapq.heappush(self.large, -heapq.heappop(self.small))
if len(self.large) > len(self.small) + 1:
heapq.heappush(self.small, -heapq.heappop(self.large))getMedian() 返回什么?要注意符号和长度奇偶性
中位数定义没变,但实现上必须区分奇偶:堆总长为奇数时,中位数就是 large[0](因为 large 多存一个);为偶数时,是 (-small[0] + large[0]) / 2。这里最容易错的是忘记对 small[0] 取负——它存的是负值,直接用会得到错误的大数。
- 不要写
self.small[0] + self.large[0],那是负数加正数 - 不要在
addNum里反复调用len()判断奇偶,只在findMedian里查一次 - 如果业务只要整数中位数(如监控阈值),建议用
// 2配合int()显式截断,避免浮点误差
实际部署时 heapify 初始化和线程安全怎么处理
初始化两个空堆不用 heapq.heapify([])——空列表本身就是有效堆。但如果从已有数据批量构建(比如冷启动加载历史流),得先对原始数组做 heapq.heapify(),且注意:对 small 要先转负再建堆。
heapq 不是线程安全的。如果多个线程并发调用 addNum,必须加锁。别用 @synchronized 包整个方法,只锁堆操作那段:
import threading self._lock = threading.Lock() <p>def addNum(self, num): with self._lock:</p><h1>上面的三行堆操作放这里
真正难的不是堆逻辑,而是当数据流包含重复值、NaN 或极小/极大浮点数时,heapq 仍能稳定比较——这依赖 Python 的 float 比较规则,但遇到 inf 或 nan 仍需前置过滤。


















