每日算法 — 使用java实现数据流的中位数：双堆协同的实时统计结构

在一个持续产生数据的数据流中，如何实时获取当前所有已接收数据的中位数？这个问题看似是简单的统计问题，实则是经典的算法设计题。它考验的是如何在动态变化的数据集合中，高效地维护有序性信息。

本文将用Java实现基于双堆结构的数据流中位数求解器，讲解最大堆与最小堆的协同工作模式，分析O(log n)插入和O(1)查询的时间复杂度，并探讨其在实时监控系统中的应用。

问题定义与核心挑战

中位数的定义：将数据集合排序后，若元素个数为奇数，中位数是中间的那个数；若为偶数，中位数是中间两个数的平均值。

核心挑战在于数据是流式到达的——我们无法一次性看到所有数据，也不能每次新数据到来时都重新排序。我们需要一种数据结构，能够：
– 高效插入新元素（最好O(log n)）
– 高效查询中位数（最好O(1)）

双堆设计的核心思想

答案是使用两个堆协同工作：

• 最大堆（maxHeap）：存储较小的一半数据，堆顶是这一半中的最大值
• 最小堆（minHeap）：存储较大的一半数据，堆顶是这一半中的最小值

通过维护两个堆的大小平衡（元素个数相差不超过1），中位数可以直接从两个堆的堆顶获取：
– 若总元素为奇数：中位数是元素较多的那个堆的堆顶
– 若总元素为偶数：中位数是两个堆顶的平均值

import java.util.*;

/**
 * 数据流中位数求解器
 * 使用双堆结构：最大堆存储较小一半，最小堆存储较大一半
 */
public class MedianFinder {
    // 最大堆：存储较小的一半，Java的PriorityQueue默认是最小堆，需要反转比较器
    private final PriorityQueue<Integer> maxHeap;
    // 最小堆：存储较大的一半
    private final PriorityQueue<Integer> minHeap;

    public MedianFinder() {
        // 最大堆：大顶堆，堆顶是最大值
        maxHeap = new PriorityQueue<>(Collections.reverseOrder());
        // 最小堆：小顶堆，堆顶是最小值（默认行为）
        minHeap = new PriorityQueue<>();
    }

插入操作的平衡策略

插入新数字时，需要遵循以下规则来保持双堆的平衡性质：

1. 先放入最大堆：新数字先进入maxHeap（较小的一半）
2. 调整堆顶：将maxHeap的堆顶（较小一半的最大值）移动到minHeap
3. 检查大小平衡：如果minHeap的元素个数超过maxHeap超过1个，将minHeap堆顶移回maxHeap

为什么这样设计？因为我们希望 maxHeap.size() >= minHeap.size() 或者两者相差不超过1。通过始终先放入maxHeap再调整，我们保证了maxHeap中的元素要么等于、要么比minHeap多1个。

    /**
     * 向数据流中添加一个数字
     * 时间复杂度：O(log n)
     */
    public void addNum(int num) {
        // 步骤1：新数字先放入最大堆
        maxHeap.offer(num);

        // 步骤2：将最大堆的堆顶移动到最小堆
        // 这保证了最大堆的所有元素 <= 最小堆的所有元素
        minHeap.offer(maxHeap.poll());

        // 步骤3：平衡两个堆的大小
        // 最大堆的元素个数应该 >= 最小堆的元素个数
        if (maxHeap.size() < minHeap.size()) {
            maxHeap.offer(minHeap.poll());
        }
    }

查询中位数

    /**
     * 获取当前数据流的中位数
     * 时间复杂度：O(1)
     */
    public double findMedian() {
        if (maxHeap.size() > minHeap.size()) {
            // 奇数个元素，中位数是最大堆的堆顶
            return maxHeap.peek();
        } else {
            // 偶数个元素，中位数是两个堆顶的平均值
            return (maxHeap.peek() + minHeap.peek()) / 2.0;
        }
    }

完整可运行代码

import java.util.*;

/**
 * 数据流中位数求解器
 * 双堆协同设计：最大堆存储较小一半，最小堆存储较大一半
 */
public class MedianFinder {
    private final PriorityQueue<Integer> maxHeap; // 较小的一半
    private final PriorityQueue<Integer> minHeap; // 较大的一半

    public MedianFinder() {
        maxHeap = new PriorityQueue<>(Collections.reverseOrder());
        minHeap = new PriorityQueue<>();
    }

    public void addNum(int num) {
        maxHeap.offer(num);
        minHeap.offer(maxHeap.poll());
        if (maxHeap.size() < minHeap.size()) {
            maxHeap.offer(minHeap.poll());
        }
    }

    public double findMedian() {
        if (maxHeap.size() > minHeap.size()) {
            return maxHeap.peek();
        }
        return (maxHeap.peek() + minHeap.peek()) / 2.0;
    }

    // ========== 扩展：支持查询其他统计量 ==========

    /**
     * 获取当前数据流的大小
     */
    public int size() {
        return maxHeap.size() + minHeap.size();
    }

    /**
     * 获取当前最大值（即minHeap的堆顶，如果minHeap为空则取maxHeap堆顶）
     */
    public int getMax() {
        if (!minHeap.isEmpty()) {
            return minHeap.peek();
        }
        return maxHeap.peek();
    }

    /**
     * 获取当前最小值（即maxHeap的堆顶）
     */
    public int getMin() {
        return maxHeap.peek();
    }

    // ========== 演示 ==========

    public static void main(String[] args) {
        MedianFinder finder = new MedianFinder();
        int[] stream = {5, 2, 8, 1, 9, 3, 7, 4, 6};

        System.out.println("数据流实时中位数演示:");
        System.out.println("数据流: " + Arrays.toString(stream));
        System.out.println();

        for (int num : stream) {
            finder.addNum(num);
            System.out.printf("添加 %d 后 -> 中位数: %.1f (数据量: %d)%n",
                num, finder.findMedian(), finder.size());
        }
    }
}

运行输出：

数据流实时中位数演示:
数据流: [5, 2, 8, 1, 9, 3, 7, 4, 6]

添加 5 后 -> 中位数: 5.0 (数据量: 1)
添加 2 后 -> 中位数: 3.5 (数据量: 2)
添加 8 后 -> 中位数: 5.0 (数据量: 3)
添加 1 后 -> 中位数: 3.5 (数据量: 4)
添加 9 后 -> 中位数: 5.0 (数据量: 5)
添加 3 后 -> 中位数: 4.0 (数据量: 6)
添加 7 后 -> 中位数: 5.0 (数据量: 7)
添加 4 后 -> 中位数: 4.5 (数据量: 8)
添加 6 后 -> 中位数: 5.0 (数据量: 9)

算法复杂度分析

总空间复杂度为O(n)，因为所有元素都需要存储在两个堆中。

正确性证明

不变式：在任意时刻，maxHeap存储较小的一半元素，minHeap存储较大的一半元素，且满足：
1. maxHeap中的所有元素 ≤ minHeap中的所有元素
2. maxHeap.size() == minHeap.size() 或 maxHeap.size() == minHeap.size() + 1

证明：
– 初始状态：两个堆都为空，不变式成立。
– 插入操作：新元素先进入maxHeap，然后将maxHeap堆顶移到minHeap。此时minHeap可能比maxHeap大1，若发生则将minHeap堆顶移回maxHeap。这一系列操作保证了maxHeap中的最大值 ≤ minHeap中的最小值，且大小关系满足条件2。
– 因此每次插入后不变式都成立。

由不变式可知，当元素个数为奇数时，中位数就是元素较多的maxHeap的堆顶；为偶数时，中位数是两个堆顶的平均值。

扩展应用：实时分位数查询

双堆思想可以扩展到任意分位数的实时查询。例如，要实时查询第k大的元素，可以使用一个大小为k的最小堆，存储当前看到的前k大元素。

    /**
     * 使用大小为k的最小堆求第k大元素
     */
    public int findKthLargest(int[] nums, int k) {
        PriorityQueue<Integer> minHeap = new PriorityQueue<>();
        for (int num : nums) {
            minHeap.offer(num);
            if (minHeap.size() > k) {
                minHeap.poll(); // 移除最小的，保留前k大
            }
        }
        return minHeap.peek(); // 第k大就是堆顶
    }

总结

双堆结构是处理动态数据流统计问题的利器。它将中位数查询这一看似需要全量排序的问题，巧妙地转化为两个堆的协同维护，实现了O(log n)的插入和O(1)的查询。这种设计思想在数据库索引、实时监控系统、流式计算框架中有广泛应用。