Java Stream处理大文件排序导致内存溢出的深度解析与解决方案

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.stream.Stream;

// 假设 largeFilePath 是指向大文件的Path对象
Path largeFilePath = Path.of("path/to/your/large_file.txt");

public void processLargeFileWithSorting(Path largeFilePath) throws IOException {
    Path tempFile = null;
    try (final Stream<String> stream = Files.lines(largeFilePath, StandardCharsets.ISO_8859_1).sorted()) {
        tempFile = Files.createTempFile(null, null); // 创建临时文件
        stream.forEach(e -> {
            if (e.startsWith("aa")) { // 使用String.startsWith替代StringUtils.startsWith
                try {
                    Files.write(tempFile, (e + System.lineSeparator()).getBytes(StandardCharsets.ISO_8859_1), StandardOpenOption.APPEND);
                } catch (final IOException e1) {
                    throw new RuntimeException("写入临时文件失败", e1);
                }
            }
        });
    } catch (final Exception e) {
        // 捕获并重新抛出异常，或进行更详细的日志记录
        throw new IOException("文件处理过程中发生错误", e);
    } finally {
        // 生产环境中应考虑临时文件的清理策略
        // if (tempFile != null) {
        //     Files.deleteIfExists(tempFile);
        // }
    }
}

java -Xms1024m -Xmx2048m -XX:MaxMetaspaceSize=256m -jar your_application.jar

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;

public class ExternalSortExample {

    private static final long MAX_LINES_IN_MEMORY = 100_000; // 内存中处理的最大行数

    public Path sortLargeFile(Path largeFilePath, Path outputDirectory) throws IOException {
        List<Path> sortedChunkFiles = new ArrayList<>();
        int chunkNum = 0;

        // 1. 分块读取、内存排序并写入临时文件
        try (BufferedReader reader = Files.newBufferedReader(largeFilePath, StandardCharsets.ISO_8859_1)) {
            List<String> currentChunk = new ArrayList<>();
            String line;
            while ((line = reader.readLine()) != null) {
                currentChunk.add(line);
                if (currentChunk.size() >= MAX_LINES_IN_MEMORY) {
                    Collections.sort(currentChunk); // 内存排序
                    Path chunkFile = writeChunkToFile(currentChunk, outputDirectory, chunkNum++);
                    sortedChunkFiles.add(chunkFile);
                    currentChunk.clear();
                }
            }
            // 处理剩余的行
            if (!currentChunk.isEmpty()) {
                Collections.sort(currentChunk);
                Path chunkFile = writeChunkToFile(currentChunk, outputDirectory, chunkNum++);
                sortedChunkFiles.add(chunkFile);
            }
        }

        // 2. 多路归并
        if (sortedChunkFiles.isEmpty()) {
            return largeFilePath; // 如果文件为空或没有需要排序的行
        }
        if (sortedChunkFiles.size() == 1) {
            return sortedChunkFiles.get(0); // 如果只有一个块，直接返回
        }

        Path finalSortedFile = outputDirectory.resolve("final_sorted_output.txt");
        mergeSortedChunks(sortedChunkFiles, finalSortedFile);

        // 清理临时文件
        for (Path chunkFile : sortedChunkFiles) {
            Files.delete(chunkFile);
        }

        return finalSortedFile;
    }

    private Path writeChunkToFile(List<String> chunk, Path outputDirectory, int chunkNum) throws IOException {
        Path chunkFile = outputDirectory.resolve("chunk_" + chunkNum + ".tmp");
        try (BufferedWriter writer = Files.newBufferedWriter(chunkFile, StandardCharsets.ISO_8859_1)) {
            for (String line : chunk) {
                writer.write(line);
                writer.newLine();
            }
        }
        return chunkFile;
    }

    private void mergeSortedChunks(List<Path> chunkFiles, Path finalOutputFile) throws IOException {
        // 使用优先队列实现多路归并
        PriorityQueue<LineReader> pq = new PriorityQueue<>(
            Comparator.comparing(lr -> lr.currentLine)
        );

        List<LineReader> readers = new ArrayList<>();
        for (Path chunkFile : chunkFiles) {
            LineReader lr = new LineReader(Files.newBufferedReader(chunkFile, StandardCharsets.ISO_8859_1));
            if (lr.hasNext()) {
                pq.offer(lr);
                readers.add(lr);
            }
        }

        try (BufferedWriter writer = Files.newBufferedWriter(finalOutputFile, StandardCharsets.ISO_8859_1)) {
            while (!pq.isEmpty()) {
                LineReader smallestReader = pq.poll();
                writer.write(smallestReader.currentLine);
                writer.newLine();

                if (smallestReader.hasNext()) {
                    pq.offer(smallestReader); // 将下一个元素重新放入队列
                } else {
                    smallestReader.close(); // 关闭已读完的reader
                }
            }
        } finally {
            // 确保所有reader都被关闭
            for (LineReader reader : readers) {
                reader.close();
            }
        }
    }

    // 辅助类，用于从文件中读取行并支持优先队列
    private static class LineReader implements AutoCloseable {
        BufferedReader reader;
        String currentLine;

        public LineReader(BufferedReader reader) throws IOException {
            this.reader = reader;
            readNextLine();
        }

        public boolean hasNext() {
            return currentLine != null;
        }

        public void readNextLine() throws IOException {
            currentLine = reader.readLine();
        }

        @Override
        public void close() throws IOException {
            if (reader != null) {
                reader.close();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        // 示例用法
        Path largeFile = Paths.get("your_large_input_file.txt"); // 替换为你的大文件路径
        Path outputDir = Paths.get("./temp_sort_output"); // 临时文件和最终输出文件的目录
        Files.createDirectories(outputDir); // 确保输出目录存在

        ExternalSortExample sorter = new ExternalSortExample();
        Path sortedResult = sorter.sortLargeFile(largeFile, outputDir);
        System.out.println("排序完成，结果文件位于: " + sortedResult);
    }
}