每日算法 — 使用java实现外部排序：大文件多路归并与缓冲区管理

引言

当待排序的数据量超过内存容量时，传统的内存排序算法（如快速排序、归并排序）无法直接应用。外部排序（External Sort）是解决这一问题的标准方案，广泛应用于数据库系统、日志处理和大数据场景。其核心思想是”分而治之”：先将大文件分成多个可以装入内存的小块，分别排序后写回磁盘；然后通过多路归并将这些有序块合并为完整的有序文件。本文将详细讲解外部排序的原理，并用Java实现一个支持自定义缓冲区大小的外部排序器。

算法原理

外部排序分为两个阶段：

阶段一：初始归并段生成

1. 将大文件按内存缓冲区大小划分为若干段
2. 每次读入一段数据到内存
3. 使用内存排序算法（如快速排序）对段内数据排序
4. 将排好序的段写回临时文件

阶段二：K路归并

1. 从每个临时文件中读取一个数据块到内存缓冲区
2. 使用最小堆（优先队列）维护各文件当前最小元素
3. 每次取出堆顶元素写入输出文件
4. 从该元素所属文件读取下一个元素补充到堆中
5. 重复直到所有文件处理完毕

Java实现

文件分割与初始排序

import java.io.*;
import java.nio.file.*;
import java.util.*;

/**
 * 外部排序器
 * 适用于内存无法容纳的大文件排序
 */
public class ExternalSorter {

    // 内存缓冲区大小（元素个数）
    private int bufferSize;
    // 临时文件目录
    private Path tempDir;

    public ExternalSorter(int bufferSize) throws IOException {
        this.bufferSize = bufferSize;
        this.tempDir = Files.createTempDirectory("external_sort_");
    }

    /**
     * 执行外部排序
     * @param inputFile 输入文件路径（每行一个整数）
     * @param outputFile 输出文件路径
     */
    public void sort(String inputFile, String outputFile) throws IOException {
        // 阶段1：生成初始归并段
        List<Path> sortedRuns = createSortedRuns(inputFile);
        System.out.println("初始归并段数量: " + sortedRuns.size());

        // 阶段2：K路归并
        kWayMerge(sortedRuns, Paths.get(outputFile));

        // 清理临时文件
        cleanup(sortedRuns);
    }

    /**
     * 阶段1：创建有序归并段
     */
    private List<Path> createSortedRuns(String inputFile) throws IOException {
        List<Path> runs = new ArrayList<>();
        List<Integer> buffer = new ArrayList<>(bufferSize);

        try (BufferedReader reader = new BufferedReader(new FileReader(inputFile))) {
            String line;
            int runCount = 0;

            while ((line = reader.readLine()) != null) {
                buffer.add(Integer.parseInt(line.trim()));

                // 缓冲区满，排序并写入临时文件
                if (buffer.size() >= bufferSize) {
                    runs.add(writeSortedRun(buffer, runCount++));
                    buffer.clear();
                }
            }

            // 处理剩余数据
            if (!buffer.isEmpty()) {
                runs.add(writeSortedRun(buffer, runCount));
            }
        }

        return runs;
    }

    /**
     * 将缓冲区数据排序后写入临时文件
     */
    private Path writeSortedRun(List<Integer> buffer, int runId) throws IOException {
        Collections.sort(buffer);

        Path runFile = tempDir.resolve("run_" + runId + ".tmp");
        try (BufferedWriter writer = new BufferedWriter(new FileWriter(runFile.toFile()))) {
            for (int value : buffer) {
                writer.write(String.valueOf(value));
                writer.newLine();
            }
        }

        return runFile;
    }
}

K路归并

    /**
     * 阶段2：K路归并
     * 使用优先队列（最小堆）维护各归并段当前最小值
     */
    private void kWayMerge(List<Path> sortedRuns, Path outputFile) throws IOException {
        // 优先队列存储 (值, 归并段索引, 该归并段的BufferedReader)
        PriorityQueue<RunEntry> minHeap = new PriorityQueue<>(
            Comparator.comparingInt(e -> e.value)
        );

        // 打开所有归并段文件，读取第一个元素
        List<BufferedReader> readers = new ArrayList<>();
        for (int i = 0; i < sortedRuns.size(); i++) {
            BufferedReader reader = new BufferedReader(new FileReader(sortedRuns.get(i).toFile()));
            readers.add(reader);

            String line = reader.readLine();
            if (line != null) {
                minHeap.offer(new RunEntry(Integer.parseInt(line), i, reader));
            }
        }

        // 归并输出
        try (BufferedWriter writer = new BufferedWriter(new FileWriter(outputFile.toFile()))) {
            while (!minHeap.isEmpty()) {
                RunEntry min = minHeap.poll();
                writer.write(String.valueOf(min.value));
                writer.newLine();

                // 从同一归并段读取下一个元素
                String nextLine = min.reader.readLine();
                if (nextLine != null) {
                    minHeap.offer(new RunEntry(Integer.parseInt(nextLine), min.runIndex, min.reader));
                }
            }
        }

        // 关闭所有reader
        for (BufferedReader reader : readers) {
            reader.close();
        }
    }

    /**
     * 归并段条目：封装值、所属归并段索引和读取器
     */
    private static class RunEntry {
        int value;
        int runIndex;
        BufferedReader reader;

        RunEntry(int value, int runIndex, BufferedReader reader) {
            this.value = value;
            this.runIndex = runIndex;
            this.reader = reader;
        }
    }

    /**
     * 清理临时文件
     */
    private void cleanup(List<Path> runs) throws IOException {
        for (Path run : runs) {
            Files.deleteIfExists(run);
        }
        Files.deleteIfExists(tempDir);
    }

完整类与演示

/**
 * 外部排序演示程序
 */
public class ExternalSortDemo {

    public static void main(String[] args) throws IOException {
        String inputFile = "large_input.txt";
        String outputFile = "sorted_output.txt";

        // 生成测试数据：10000个随机整数
        System.out.println("生成测试数据...");
        generateTestData(inputFile, 10000);

        // 执行外部排序，每次内存排序1000个元素
        System.out.println("执行外部排序...");
        ExternalSorter sorter = new ExternalSorter(1000);
        long start = System.currentTimeMillis();
        sorter.sort(inputFile, outputFile);
        long end = System.currentTimeMillis();

        System.out.println("排序完成，耗时: " + (end - start) + "ms");

        // 验证排序结果
        System.out.println("验证排序结果...");
        boolean isSorted = verifySorted(outputFile);
        System.out.println("排序正确: " + isSorted);

        // 显示前20个结果
        System.out.println("\n前20个排序结果:");
        showFirstN(outputFile, 20);
    }

    /**
     * 生成随机测试数据
     */
    static void generateTestData(String filename, int count) throws IOException {
        Random random = new Random(42);  // 固定种子以便复现
        try (BufferedWriter writer = new BufferedWriter(new FileWriter(filename))) {
            for (int i = 0; i < count; i++) {
                writer.write(String.valueOf(random.nextInt(100000)));
                writer.newLine();
            }
        }
    }

    /**
     * 验证文件是否已排序
     */
    static boolean verifySorted(String filename) throws IOException {
        try (BufferedReader reader = new BufferedReader(new FileReader(filename))) {
            String line;
            int prev = Integer.MIN_VALUE;
            while ((line = reader.readLine()) != null) {
                int current = Integer.parseInt(line.trim());
                if (current < prev) {
                    return false;
                }
                prev = current;
            }
        }
        return true;
    }

    /**
     * 显示文件前N行
     */
    static void showFirstN(String filename, int n) throws IOException {
        try (BufferedReader reader = new BufferedReader(new FileReader(filename))) {
            String line;
            int count = 0;
            while ((line = reader.readLine()) != null && count < n) {
                System.out.print(line + " ");
                count++;
                if (count % 10 == 0) System.out.println();
            }
        }
    }
}

复杂度分析

关键优化

1. 增大缓冲区：减少初始归并段数量，降低归并趟数
2. 提高归并路数：使用更大的K值，减少归并树高度
3. 并行读取：利用多线程同时读取多个归并段
4. 缓冲区预读：使用操作系统页缓存减少实际磁盘I/O
5. 置换-选择排序：生成更长的初始归并段，进一步减少段数

实际应用场景

1. 数据库排序操作：Oracle、MySQL处理大数据量ORDER BY
2. 日志分析：按时间戳排序TB级日志文件
3. 搜索引擎：对海量网页文档进行排序和去重
4. 基因组学：DNA序列排序和比对
5. MapReduce：Hadoop中Map阶段的排序和Reduce阶段的归并

总结

外部排序是处理超大规模数据排序的基石算法。本文通过Java实现展示了从文件分割、内存排序到K路归并的完整流程。理解外部排序的关键在于认识到磁盘I/O是主要性能瓶颈，因此算法设计的核心目标是最小化磁盘访问次数。在实际工程中，缓冲区大小和归并路数的调优对性能影响巨大。