Top K 问题：最小堆、分区法与分布式解法 ⭐⭐⭐

/**
 * 求 Top K 大数（最小堆）
 * 时间：O(N log K)
 * 空间：O(K)
 */
public List<Integer> topK(int[] arr, int k) {
    if (k <= 0) return Collections.emptyList();

    // Java 的 PriorityQueue 默认是最小堆
    PriorityQueue<Integer> minHeap = new PriorityQueue<>(k);

    for (int num : arr) {
        if (minHeap.size() < k) {
            // 堆未满，直接加入
            minHeap.offer(num);
        } else if (num > minHeap.peek()) {
            // 比堆顶大，替换
            minHeap.poll();    // 移除堆顶（最小值）
            minHeap.offer(num); // 加入新元素
        }
        // 比堆顶小 → 不处理，直接跳过
    }

    // 堆内即为 Top K（无序）
    return new ArrayList<>(minHeap);
}

/**
 * 求 Top K 小数（最大堆）
 */
public List<Integer> topKMin(int[] arr, int k) {
    if (k <= 0) return Collections.emptyList();

    // 最大堆：传入自定义比较器
    PriorityQueue<Integer> maxHeap = new PriorityQueue<>(
        (a, b) -> b - a  // 最大堆
    );

    for (int num : arr) {
        if (maxHeap.size() < k) {
            maxHeap.offer(num);
        } else if (num < maxHeap.peek()) {
            // 比堆顶小才替换（找最小的 K 个）
            maxHeap.poll();
            maxHeap.offer(num);
        }
    }

    return new ArrayList<>(maxHeap);
}

/**
 * Top K 小数的另一种写法（Java 泛型版）
 */
public <T> List<T> topK(List<T> arr, int k, Comparator<T> comp) {
    PriorityQueue<T> heap = new PriorityQueue<>(comp);
    for (T item : arr) {
        if (heap.size() < k) {
            heap.offer(item);
        } else if (comp.compare(item, heap.peek()) > 0) {
            // item 比堆顶大（对于最大堆，意味着 item 更"大"）
            heap.poll();
            heap.offer(item);
        }
    }
    return new ArrayList<>(heap);
}

/**
 * 快速选择：原地分区找 Top K 大
 * 时间复杂度：O(N)（均摊）
 * 空间复杂度：O(1)
 */
public List<Integer> topKQuickSelect(int[] arr, int k) {
    if (k <= 0 || k > arr.length) {
        return Collections.emptyList();
    }

    int[] nums = arr.clone(); // 克隆避免修改原数组
    int targetIdx = k - 1;    // Top K 大的第 K 个元素的目标索引

    int lo = 0, hi = nums.length - 1;
    while (lo <= hi) {
        int pivotIdx = partition(nums, lo, hi);

        if (pivotIdx == targetIdx) {
            // 找到目标位置
            break;
        } else if (pivotIdx < targetIdx) {
            // pivot 偏左，目标在右半部分
            lo = pivotIdx + 1;
        } else {
            // pivot 偏右，目标在左半部分
            hi = pivotIdx - 1;
        }
    }

    // 收集 [k, end] 的数即为 Top K（这部分都 >= 第 K 大的数）
    // 注意：此时 k 位置是第 K 大的数，左边可能有小于它的
    List<Integer> result = new ArrayList<>();
    for (int i = k - 1; i < nums.length; i++) {
        result.add(nums[i]);
    }
    // 对结果排序（可选）
    result.sort((a, b) -> b - a);
    return result;
}

/**
 * 分区：选 hi 为 pivot，比 pivot 大的放左边，小的放右边
 * 返回 pivot 最终位置
 */
private int partition(int[] nums, int lo, int hi) {
    int pivot = nums[hi];
    int i = lo - 1; // 划分边界

    for (int j = lo; j < hi; j++) {
        if (nums[j] >= pivot) { // 大于等于 pivot 放左边
            swap(nums, ++i, j);
        }
    }
    // pivot 放到分界线
    swap(nums, i + 1, hi);
    return i + 1;
}

private void swap(int[] nums, int a, int b) {
    int tmp = nums[a];
    nums[a] = nums[b];
    nums[b] = tmp;
}

// Map 阶段：单机 Top K
public class TopKMapper {
    private static final int K = 1000;

    public void map(String filename, Iterator<Long> records) {
        PriorityQueue<Long> localTopK = new PriorityQueue<>(K); // 最小堆

        while (records.hasNext()) {
            long num = records.next();
            if (localTopK.size() < K) {
                localTopK.offer(num);
            } else if (num > localTopK.peek()) {
                localTopK.poll();
                localTopK.offer(num);
            }
        }

        // 输出本机 Top K
        for (Long num : localTopK) {
            collect(num); // 输出到文件
        }
    }
}

// Reduce 阶段：全局 Top K
public class TopKReducer {
    private static final int K = 1000;

    public void reduce(Iterator<Long> allNumbers) {
        PriorityQueue<Long> globalTopK = new PriorityQueue<>(K);

        while (allNumbers.hasNext()) {
            long num = allNumbers.next();
            if (globalTopK.size() < K) {
                globalTopK.offer(num);
            } else if (num > globalTopK.peek()) {
                globalTopK.poll();
                globalTopK.offer(num);
            }
        }

        // 输出最终 Top K
        for (Long num : globalTopK) {
            output(num);
        }
    }
}

/**
 * 数据流滑动窗口 Top K
 */
public class SlidingWindowTopK {
    private final int windowSize;
    private final int k;
    private final LinkedList<Long> window = new LinkedList<>();
    private final PriorityQueue<Long> topK = new PriorityQueue<>(Comparator.reverseOrder()); // 最大堆

    public SlidingWindowTopK(int windowSize, int k) {
        this.windowSize = windowSize;
        this.k = k;
    }

    public void add(long num) {
        // 1. 加入窗口
        window.addLast(num);
        if (window.size() > windowSize) {
            long removed = window.removeFirst();
            // 从 Top K 中移除（需要重建堆，简化处理）
            topK.remove(removed);
        }

        // 2. 更新 Top K
        topK.add(num);

        // 3. 如果 Top K 超过 K，移除多余
        while (topK.size() > k) {
            // 对于最大堆，poll 的是最大的，但我们需要保留 K 个最大的
            // 所以最大堆的逻辑需要反过来
            topK.poll(); // 移除最大的... 这不对
        }

        // 修正：窗口内 Top K = 最大堆，直接保留 K 个最大
        // 实际上用最大堆更合适
    }
}