每日算法 — 使用java实现败者树：外排序k路归并的锦标赛选择优化

引言

败者树（Loser Tree）是外排序中k路归并的核心数据结构，它与胜者树（Winner Tree）互补，通过记录”败者”而非”胜者”来优化比较次数。在k路归并中，每轮需要选出k个有序段中的最小值，败者树将这一选择操作的比较次数从O(k)优化到O(log k)。本文将用Java实现完整的败者树，并展示其在外排序k路归并中的实际应用。

算法核心思想

胜者树与败者树的对比

胜者树：每个内部节点记录左右子树比较的胜者。当胜者被取出后，新元素需要从叶子向上更新，每层都需要读取兄弟节点（败者）进行比较。

败者树：每个内部节点记录比较的败者，胜者继续向上参与比较，根节点的父节点额外记录最终胜者。更新时只需与新进入的兄弟节点比较，无需访问原来的胜者。

败者树的核心优势

在k路归并中，每轮取走最小值后需要补充新元素。败者树更新时，新元素从叶子向上，每层只需与当前节点存储的败者比较，而胜者树需要读取兄弟节点的值。败者树因此减少了数组访问次数，在实践中更快。

树的结构

对于k个叶子节点，败者树使用大小为k的内部节点数组（索引1到k-1）和一个独立的winner变量存储全局最小值。叶子节点对应k个归并段的当前元素。

完整Java实现

import java.util.*;

/**
 * 败者树（Loser Tree）实现
 * 用于k路归并中高效选择最小值
 * @param <T> 元素类型，必须实现Comparable接口
 */
public class LoserTree<T extends Comparable<T>> {

    private final Object[] leaves;      // 叶子节点，存储各段的当前元素
    private final int[] loserTree;      // 败者树内部节点，存储败者索引
    private int winner;                 // 全局胜者索引（最终最小值所在叶子）
    private final int k;                // 路数

    /**
     * 构造败者树
     * @param initialValues k个初始值，每个代表一路的当前元素
     */
    @SuppressWarnings("unchecked")
    public LoserTree(T[] initialValues) {
        this.k = initialValues.length;
        this.leaves = new Object[k];
        this.loserTree = new int[k];  // 内部节点数等于叶子数（简化实现，实际用k-1个）

        // 初始化叶子
        for (int i = 0; i < k; i++) {
            leaves[i] = initialValues[i];
        }

        // 构建败者树
        buildTree();
    }

    /**
     * 比较两个叶子节点的值
     * @return true if leaves[a] < leaves[b]（a胜）
     */
    @SuppressWarnings("unchecked")
    private boolean lessThan(int a, int b) {
        T valA = (T) leaves[a];
        T valB = (T) leaves[b];
        if (valA == null) return false;  // null表示该路已空，视为无穷大
        if (valB == null) return true;
        return valA.compareTo(valB) < 0;
    }

    /**
     * 构建败者树
     * 初始化所有内部节点，通过锦标赛方式逐层确定败者
     */
    private void buildTree() {
        // 先让所有叶子参与一轮比赛，确定初始败者
        // 简化实现：使用完全二叉树结构，内部节点索引1..k-1

        // 初始化winner为0号叶子
        winner = 0;

        // 从1号叶子开始，逐个点与当前winner比赛
        for (int i = 1; i < k; i++) {
            // i与当前winner比赛
            if (lessThan(i, winner)) {
                // i胜，原winner成为败者，存储在当前层
                loserTree[i] = winner;
                winner = i;
            } else {
                // winner胜，i成为败者
                loserTree[i] = i;
            }
        }

        // 上面的简化方法不完全正确，下面是标准的完全二叉树构建
        // 重新用标准的完全二叉树方式构建
        Arrays.fill(loserTree, -1);

        // 找到第一个大于等于k的2的幂
        int treeSize = 1;
        while (treeSize < k) treeSize <<= 1;

        // 使用扩展数组方式：内部节点从1开始编号
        // 叶子在树中的位置为 treeSize/2 + i
        int[] internal = new int[treeSize];  // 内部节点存储败者索引
        Arrays.fill(internal, -1);

        // 初始化所有叶子位置
        int leafStart = treeSize / 2;

        // 自底向上构建
        for (int i = 0; i < k; i++) {
            int pos = leafStart + i;
            int current = i;

            while (pos > 1) {
                pos >>= 1;  // 父节点
                if (internal[pos] == -1) {
                    // 该节点为空，当前节点暂时成为胜者继续向上
                    internal[pos] = current;
                    break;
                } else {
                    // 与存储的节点比赛
                    int opponent = internal[pos];
                    if (lessThan(current, opponent)) {
                        // current胜，opponent成为败者留在该节点
                        internal[pos] = opponent;
                        // current继续向上
                    } else {
                        // opponent胜，current成为败者留在该节点
                        internal[pos] = current;
                        current = opponent;
                        // opponent继续向上
                    }
                }
            }

            if (pos == 1) {
                winner = current;  // 最终胜者
            }
        }

        // 将内部节点复制到loserTree（适配外部接口）
        System.arraycopy(internal, 0, loserTree, 0, Math.min(k, internal.length));
    }

    /**
     * 获取当前最小值（胜者）
     */
    @SuppressWarnings("unchecked")
    public T getWinner() {
        return (T) leaves[winner];
    }

    /**
     * 获取胜者索引
     */
    public int getWinnerIndex() {
        return winner;
    }

    /**
     * 替换胜者叶子为新值，并重新调整败者树
     * @param newValue 新值，null表示该路已结束
     */
    public void replaceWinner(T newValue) {
        leaves[winner] = newValue;

        // 找到叶子在完全二叉树中的位置
        int treeSize = 1;
        while (treeSize < k) treeSize <<= 1;
        int leafStart = treeSize / 2;

        int pos = leafStart + winner;
        int current = winner;

        while (pos > 1) {
            pos >>= 1;
            int opponent = loserTree[pos];

            if (opponent == -1) {
                // 只有一个孩子
                continue;
            }

            if (lessThan(current, opponent)) {
                // current胜，opponent保持为败者
                // current继续向上
            } else {
                // opponent胜，current成为新的败者
                loserTree[pos] = current;
                current = opponent;
            }
        }

        winner = current;
    }

    /**
     * 检查是否所有路都已结束
     */
    public boolean isEmpty() {
        return leaves[winner] == null;
    }

    /**
     * 打印败者树结构（调试用）
     */
    public void printTree() {
        System.out.println("Loser Tree (winner=" + winner + "):");
        System.out.println("Winner value: " + leaves[winner]);
        for (int i = 1; i < k && i < loserTree.length; i++) {
            System.out.println("  Node " + i + ": loser=" + loserTree[i] +
                " value=" + leaves[loserTree[i]]);
        }
    }

    // ==================== k路归并应用 ====================

    /**
     * k路归并：将k个有序数组合并为一个有序数组
     * @param arrays k个有序数组
     * @return 合并后的有序数组
     */
    @SafeVarargs
    public static <T extends Comparable<T>> List<T> kWayMerge(T[]... arrays) {
        int k = arrays.length;
        if (k == 0) return new ArrayList<>();

        // 每个数组的当前索引
        int[] indices = new int[k];

        // 构建初始叶子值
        @SuppressWarnings("unchecked")
        T[] initialValues = (T[]) new Comparable[k];
        for (int i = 0; i < k; i++) {
            initialValues[i] = arrays[i].length > 0 ? arrays[i][0] : null;
        }

        LoserTree<T> tree = new LoserTree<>(initialValues);
        List<T> result = new ArrayList<>();

        while (!tree.isEmpty()) {
            T minVal = tree.getWinner();
            int minIdx = tree.getWinnerIndex();
            result.add(minVal);

            // 推进该路的索引
            indices[minIdx]++;
            T nextVal = indices[minIdx] < arrays[minIdx].length
                ? arrays[minIdx][indices[minIdx]]
                : null;

            tree.replaceWinner(nextVal);
        }

        return result;
    }

    /**
     * 主程序：演示败者树与k路归并
     */
    public static void main(String[] args) {
        // 测试1：基本败者树操作
        System.out.println("=== 基本败者树测试 ===");
        Integer[] initial = {3, 1, 4, 1, 5};
        LoserTree<Integer> tree = new LoserTree<>(initial);
        tree.printTree();

        System.out.println("\n逐轮取出最小值:");
        int round = 1;
        while (!tree.isEmpty()) {
            System.out.println("Round " + round + ": min=" + tree.getWinner() +
                " from leaf " + tree.getWinnerIndex());
            tree.replaceWinner(null);  // 模拟取走
            round++;
        }

        // 测试2：k路归并
        System.out.println("\n=== k路归并测试 ===");
        Integer[] arr1 = {1, 4, 7, 10};
        Integer[] arr2 = {2, 5, 8, 11};
        Integer[] arr3 = {3, 6, 9, 12};

        List<Integer> merged = kWayMerge(arr1, arr2, arr3);
        System.out.println("归并结果: " + merged);

        // 测试3：大规模性能测试
        System.out.println("\n=== 大规模性能测试 ===");
        int k = 100;           // 100路归并
        int perArray = 10000;  // 每路10000个元素

        @SuppressWarnings("unchecked")
        Integer[][] largeArrays = new Integer[k][];
        Random rand = new Random(42);

        for (int i = 0; i < k; i++) {
            largeArrays[i] = new Integer[perArray];
            int start = rand.nextInt(1000);
            for (int j = 0; j < perArray; j++) {
                largeArrays[i][j] = start + j * 2 + rand.nextInt(2);
            }
            Arrays.sort(largeArrays[i]);
        }

        long start = System.currentTimeMillis();
        List<Integer> largeMerged = kWayMerge(largeArrays);
        long elapsed = System.currentTimeMillis() - start;

        System.out.println("k=" + k + ", 每路" + perArray + "元素");
        System.out.println("归并后总元素: " + largeMerged.size());
        System.out.println("败者树归并耗时: " + elapsed + "ms");

        // 验证有序性
        boolean sorted = true;
        for (int i = 1; i < largeMerged.size(); i++) {
            if (largeMerged.get(i) < largeMerged.get(i - 1)) {
                sorted = false;
                break;
            }
        }
        System.out.println("结果有序: " + sorted);
    }
}

关键算法解析

1. 败者树的存储结构

败者树采用完全二叉树的紧凑存储。对于k个叶子，找到不小于k的最小2的幂treeSize，叶子从treeSize/2开始存放，内部节点从1到treeSize/2-1。每个内部节点存储该次比较的败者索引。

2. 构建过程

初始构建时，每个叶子自底向上参与比较。若某内部节点为空，当前节点暂时成为胜者继续向上；若已有败者存储，则两者比较，败者留在该节点，胜者继续向上。最终到达根节点的即为全局胜者。

3. 更新过程

当全局胜者被取走后，该叶子更新为新值（或null），然后同样自底向上比较。由于存储的是败者，新值只需与每层的败者比较即可，无需访问胜者。这正是败者树比胜者树高效的根源。

复杂度分析

相比朴素k路归并的O(n×k)，败者树将时间复杂度优化到O(n log k)，在k较大时优势明显。

总结

败者树是外排序k路归并的经典优化数据结构。相比胜者树，它通过存储败者减少了更新时的数组访问次数，在实际运行中更快。

核心设计要点：
1. 存储败者而非胜者：更新时只需读取当前节点存储的败者进行比较
2. 完全二叉树紧凑存储：利用数组模拟树结构，无指针开销
3. O(log k)选择效率：与堆相同的选择效率，但更利于k路归并的连续更新场景

本文实现的败者树可直接应用于外排序、多路流合并等场景，是处理大规模有序数据合并的必备工具。