java - 将文件列表读取为 Java 8 流

Question

我有一个(可能很长)二进制文件列表，我想懒惰地阅读它们。将有太多文件加载到内存中。我目前正在使用 FileChannel.map() 将它们作为 MappedByteBuffer 读取，但这可能不是必需的。我希望方法 readBinaryFiles(...) 返回一个 Java 8 Stream，这样我就可以在访问文件列表时延迟加载它们。

    public List<FileDataMetaData> readBinaryFiles(
    List<File> files, 
    int numDataPoints, 
    int dataPacketSize )
    throws
    IOException {

    List<FileDataMetaData> fmdList = new ArrayList<FileDataMetaData>();

    IOException lastException = null;
    for (File f: files) {

        try {
            FileDataMetaData fmd = readRawFile(f, numDataPoints, dataPacketSize);
            fmdList.add(fmd);
        } catch (IOException e) {
            logger.error("", e);
            lastException = e;
        }
    }

    if (null != lastException)
        throw lastException;

    return fmdList;
}


//  The List<DataPacket> returned will be in the same order as in the file.
public FileDataMetaData readRawFile(File file, int numDataPoints, int dataPacketSize) throws IOException {

    FileDataMetaData fmd;
    FileChannel fileChannel = null;
    try {
        fileChannel = new RandomAccessFile(file, "r").getChannel();
        long fileSz = fileChannel.size();
        ByteBuffer bbRead = ByteBuffer.allocate((int) fileSz);
        MappedByteBuffer buffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileSz);

        buffer.get(bbRead.array());
        List<DataPacket> dataPacketList = new ArrayList<DataPacket>();

        while (bbRead.hasRemaining()) {

            int channelId = bbRead.getInt();
            long timestamp = bbRead.getLong();
            int[] data = new int[numDataPoints];
            for (int i=0; i<numDataPoints; i++) 
                data[i] = bbRead.getInt();

            DataPacket dp = new DataPacket(channelId, timestamp, data);
            dataPacketList.add(dp);
        }

        fmd = new FileDataMetaData(file.getCanonicalPath(), fileSz, dataPacketList);

    } catch (IOException e) {
        logger.error("", e);
        throw e;
    } finally {
        if (null != fileChannel) {
            try {
                fileChannel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    return fmd;
}

从 readBinaryFiles(...) 返回 fmdList.Stream() 不会完成这个，因为文件内容已经被读入内存，我赢了做不到。

将多个文件的内容作为流读取的其他方法依赖于使用 Files.lines()，但我需要读取二进制文件。

我愿意在 Scala 或 golang 中执行此操作，如果这些语言比 Java 能更好地支持此用例的话。

如果有任何关于如何懒惰地读取多个二进制文件内容的指示，我将不胜感激。

Answer 1

在读取整个文件以构建 FileDataMetaData 实例时，在 文件中的读取不可能有惰性。您需要对该类进行大量重构，以便能够构造 FileDataMetaData 的实例，而无需读取整个文件。

但是，该代码中有几处需要清理，甚至特定于 Java 7 而不是 Java 8，即您不再需要 RandomAccessFile 绕道打开 channel ，并且有try-with-resources以确保正确关闭。请进一步注意，您使用内存映射没有任何意义。在映射文件后将全部内容复制到堆 ByteBuffer 中时，没有什么偷懒的。这与在 channel 上使用堆 ByteBuffer 调用 read 时发生的情况完全相同，只是 JRE 可以在 read 情况下重用缓冲区.

为了让系统管理页面，你必须从映射的字节缓冲区中读取。根据系统的不同，这可能仍然不比将小块重复读入堆字节缓冲区更好。

public FileDataMetaData readRawFile(
    File file, int numDataPoints, int dataPacketSize) throws IOException {

    try(FileChannel fileChannel=FileChannel.open(file.toPath(), StandardOpenOption.READ)) {
        long fileSz = fileChannel.size();
        MappedByteBuffer bbRead=fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileSz);
        List<DataPacket> dataPacketList = new ArrayList<>();
        while(bbRead.hasRemaining()) {
            int channelId = bbRead.getInt();
            long timestamp = bbRead.getLong();
            int[] data = new int[numDataPoints];
            for (int i=0; i<numDataPoints; i++) 
                data[i] = bbRead.getInt();
            dataPacketList.add(new DataPacket(channelId, timestamp, data));
        }
        return new FileDataMetaData(file.getCanonicalPath(), fileSz, dataPacketList);
    } catch (IOException e) {
        logger.error("", e);
        throw e;
    }
}

基于此方法构建 Stream 非常简单，只需处理已检查的异常:

public Stream<FileDataMetaData> readBinaryFiles(
    List<File> files, int numDataPoints, int dataPacketSize) throws IOException {
    return files.stream().map(f -> {
        try {
            return readRawFile(f, numDataPoints, dataPacketSize);
        } catch (IOException e) {
            logger.error("", e);
            throw new UncheckedIOException(e);
        }
    });
}