用Java处理大文件

最近，我不得不处理一组包含逐笔历史汇率市场数据的文件，并很快意识到使用传统的InputStream都无法将它们读取到内存中，因为每个文件的大小都超过4 GB。 Emacs甚至无法打开它们。

在这种特殊情况下，我可以编写一个简单的bash脚本，将文件分成小块，然后像往常一样读取它们。但是我不希望这样，因为二进制格式会使这种方法无效。

因此，正确处理此问题的方法是使用内存映射文件逐步处理数据区域。内存映射文件的优点在于它们不消耗虚拟内存或分页空间，因为它们由磁盘上的文件数据支持。

Okey，让我们看一下这些文件并提取一些数据。似乎它们包含带有逗号分隔字段的ASCII文本行。

格式： [currency-pair],[timestamp],[bid-price],[ask-price]

例如： EUR/USD,20120102 00:01:30.420,1.29451,1.2949

公平地说，我可以为该格式编写程序。但是读取和解析文件是正交的概念。因此，让我们退后一步，考虑一下可以在将来遇到类似问题时可以重用的通用设计。

问题归结为对一组以无限长字节数组编码的条目进行增量解码，而不会耗尽内存。示例格式以逗号/行分隔的文本编码的事实与一般解决方案无关，因此很明显，需要解码器接口才能处理不同的格式。

同样，在处理完整个文件之前，无法解析每个条目并将其保留在内存中，因此我们需要一种方法来逐步移交可以在其他位置（磁盘或网络）写入的条目块，然后再进行垃圾回收。迭代器是处理此要求的很好的抽象方法，因为它们的行为就像游标一样，这正是重点。每次迭代都会转发文件指针，然后让我们对数据进行处理。

所以首先是Decoder接口。这个想法是从MappedByteBuffer增量解码对象，或者如果缓冲区中没有对象，则返回null。

public interface Decoder<T> {public T decode(ByteBuffer buffer);
}

然后是实现Iterable的FileReader 。每次迭代将处理下一个4096字节的数据，并使用Decoder将其Decoder为对象列表。请注意， FileReader接受文件列表，这很不错，因为它允许遍历数据而无需担心跨文件聚合。顺便说一下，对于较大的文件，4096个字节的块可能会有点小。

public class FileReader implements Iterable<List<T>> {private static final long CHUNK_SIZE = 4096;private final Decoder<T> decoder;private Iterator<File> files;private FileReader(Decoder<T> decoder, File... files) {this(decoder, Arrays.asList(files));}private FileReader(Decoder<T> decoder, List<File> files) {this.files = files.iterator();this.decoder = decoder;}public static <T> FileReader<T> create(Decoder<T> decoder, List<File> files) {return new FileReader<T>(decoder, files);}public static <T> FileReader<T> create(Decoder<T> decoder, File... files) {return new FileReader<T>(decoder, files);}@Overridepublic Iterator<List<T>> iterator() {return new Iterator<List<T>>() {private List<T> entries;private long chunkPos = 0;private MappedByteBuffer buffer;private FileChannel channel;@Overridepublic boolean hasNext() {if (buffer == null || !buffer.hasRemaining()) {buffer = nextBuffer(chunkPos);if (buffer == null) {return false;}}T result = null;while ((result = decoder.decode(buffer)) != null) {if (entries == null) {entries = new ArrayList<T>();}entries.add(result);}// set next MappedByteBuffer chunkchunkPos += buffer.position();buffer = null;if (entries != null) {return true;} else {Closeables.closeQuietly(channel);return false;}}private MappedByteBuffer nextBuffer(long position) {try {if (channel == null || channel.size() == position) {if (channel != null) {Closeables.closeQuietly(channel);channel = null;}if (files.hasNext()) {File file = files.next();channel = new RandomAccessFile(file, "r").getChannel();chunkPos = 0;position = 0;} else {return null;}}long chunkSize = CHUNK_SIZE;if (channel.size() - position < chunkSize) {chunkSize = channel.size() - position;}return channel.map(FileChannel.MapMode.READ_ONLY, chunkPos, chunkSize);} catch (IOException e) {Closeables.closeQuietly(channel);throw new RuntimeException(e);}}@Overridepublic List<T> next() {List<T> res = entries;entries = null;return res;}@Overridepublic void remove() {throw new UnsupportedOperationException();}};}
}

下一个任务是编写一个Decoder ，我决定为任何逗号分隔的文本文件格式实现一个通用的TextRowDecoder ，接受每行的字段数和一个字段定界符，并返回一个字节数组数组。然后， TextRowDecoder可以由可能处理不同字符集的格式特定的解码器重用。

public class TextRowDecoder implements Decoder<byte[][]> {private static final byte LF = 10;private final int numFields;private final byte delimiter;public TextRowDecoder(int numFields, byte delimiter) {this.numFields = numFields;this.delimiter = delimiter;}@Overridepublic byte[][] decode(ByteBuffer buffer) {int lineStartPos = buffer.position();int limit = buffer.limit();while (buffer.hasRemaining()) {byte b = buffer.get();if (b == LF) { // reached line feed so parse lineint lineEndPos = buffer.position();// set positions for one row duplicationif (buffer.limit() < lineEndPos + 1) {buffer.position(lineStartPos).limit(lineEndPos);} else {buffer.position(lineStartPos).limit(lineEndPos + 1);}byte[][] entry = parseRow(buffer.duplicate());if (entry != null) {// reset main bufferbuffer.position(lineEndPos);buffer.limit(limit);// set start after LFlineStartPos = lineEndPos;}return entry;}}buffer.position(lineStartPos);return null;}public byte[][] parseRow(ByteBuffer buffer) {int fieldStartPos = buffer.position();int fieldEndPos = 0;int fieldNumber = 0;byte[][] fields = new byte[numFields][];while (buffer.hasRemaining()) {byte b = buffer.get();if (b == delimiter || b == LF) {fieldEndPos = buffer.position();// save limitint limit = buffer.limit();// set positions for one row duplicationbuffer.position(fieldStartPos).limit(fieldEndPos);fields[fieldNumber] = parseField(buffer.duplicate(), fieldNumber, fieldEndPos - fieldStartPos - 1);fieldNumber++;// reset main bufferbuffer.position(fieldEndPos);buffer.limit(limit);// set start after LFfieldStartPos = fieldEndPos;}if (fieldNumber == numFields) {return fields;}}return null;}private byte[] parseField(ByteBuffer buffer, int pos, int length) {byte[] field = new byte[length];for (int i = 0; i < field.length; i++) {field[i] = buffer.get();}return field;}
}

这就是文件的处理方式。每个列表包含从单个缓冲区解码的元素，每个元素都是由TextRowDecoder指定的字节数组的数组。

TextRowDecoder decoder = new TextRowDecoder(4, comma);
FileReader<byte[][]> reader = FileReader.create(decoder, file.listFiles());
for (List<byte[][]> chunk : reader) {// do something with each chunk
}

我们可以在这里停下来，但还有其他要求。每行都包含一个时间戳记，并且必须按时间段而不是按天或按小时对缓冲区进行分组。我仍然想遍历每个批次，因此立即的反应是为FileReader创建一个Iterable包装器，以实现此行为。另外一个细节是，每个元素必须通过实现PeriodEntries Timestamped接口（此处未显示）为PeriodEntries提供其时间戳。

public class PeriodEntries<T extends Timestamped> implements Iterable<List<T>> {private final Iterator<List<T extends Timestamped>> entriesIt;private final long interval;private PeriodEntries(Iterable<List<T>> entriesIt, long interval) {this.entriesIt = entriesIt.iterator();this.interval = interval;}public static <T extends Timestamped> PeriodEntries<T> create(Iterable<List<T>> entriesIt, long interval) {return new PeriodEntries<T>(entriesIt, interval);}@Overridepublic Iterator<List<T extends Timestamped>> iterator() {return new Iterator<List<T>>() {private Queue<List<T>> queue = new LinkedList<List<T>>();private long previous;private Iterator<T> entryIt;@Overridepublic boolean hasNext() {if (!advanceEntries()) {return false;}T entry =  entryIt.next();long time = normalizeInterval(entry);if (previous == 0) {previous = time;}if (queue.peek() == null) {List<T> group = new ArrayList<T>();queue.add(group);}while (previous == time) {queue.peek().add(entry);if (!advanceEntries()) {break;}entry = entryIt.next();time = normalizeInterval(entry);}previous = time;List<T> result = queue.peek();if (result == null || result.isEmpty()) {return false;}return true;}private boolean advanceEntries() {// if there are no rows leftif (entryIt == null || !entryIt.hasNext()) {// try get more rows if possibleif (entriesIt.hasNext()) {entryIt = entriesIt.next().iterator();return true;} else {// no more rowsreturn false;}}return true;}private long normalizeInterval(Timestamped entry) {long time = entry.getTime();int utcOffset = TimeZone.getDefault().getOffset(time);long utcTime = time + utcOffset;long elapsed = utcTime % interval;return time - elapsed;}@Overridepublic List<T> next() {return queue.poll();}@Overridepublic void remove() {throw new UnsupportedOperationException();}};}
}

引入此功能后，最终处理代码并没有太大变化，只有一个干净紧凑的for循环，不必关心跨文件，缓冲区和句点对元素进行分组。 PeriodEntries也足够灵活，可以管理间隔上的任何长度。

TrueFxDecoder decoder = new TrueFxDecoder();
FileReader<TrueFxData> reader = FileReader.create(decoder, file.listFiles());
long periodLength = TimeUnit.DAYS.toMillis(1);
PeriodEntries<TrueFxData> periods = PeriodEntries.create(reader, periodLength);for (List<TrueFxData> entries : periods) {// data for each dayfor (TrueFxData entry : entries) {// process each entry}
}

正如您可能意识到的那样，不可能用集合来解决这个问题。选择迭代器是一项关键的设计决策，它能够解析TB级的数据而不会占用太多的堆空间。

参考： Deephacks博客上的JCG合作伙伴 Kristoffer Sjogren 使用Java处理大型文件。