用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處理大型文件。