文章目录
最近要处理一套存储历史实时数据的大文件fx market data，使用传统的InputStream不能够将它们读取到内存，因为每一个文件都超过了4G。甚至编辑器都不能够打开这些文件。
处理这个问题的方式通常就是使用内存映射文件递增地处理区域的数据。关于内存映射文件的一个好处就是它们不会使用虚拟内存和换页空间，因为它们是从磁盘上的文件返回来的数据。
这些文件和额外的一些数据，似乎它们使用逗号分隔的字段包含ASCII文本行。
格式：[currency-pair],[timestamp],[bid-price],[ask-price]
例子：EUR/USD,20120102 00:01:30.420,1.29451,1.2949
可以为这种格式去写一个程序，但是，读取文件和解析文件是无关的概念。让我们退一步来想一个通用的设计，当在将来面临相似的问题时这个设计可以被重复利用。
这个问题可以归结为递增地解码一个已经在无限长的数组中被编码的记录，并且没有耗尽内存。实际上，以逗号分割的示例格式编码与通常的解决方案是不相关的。所以，很明显需要一个解码器来处理不同的格式。
再来看，知道整个文件处理完成，每一条记录都不能被解析并保存在内存中，所以我们需要一种方式来转移记录，在它们成为垃圾被回收之前可以被写到其他地方，例如磁盘或者网络。
迭代器是处理这个需求的很好的抽象，因为它们就像游标一样，可以正确的指向某个位置。每一次迭代都可以转发文件指针，并且可以让我们使用数据做其他的事情。
首先来写一个Decoder 接口，递增地把对象从MappedByteBuffer中解码，如果buffer中没有对象，则返回null。
public interface Decoder<T> {
public T decode(ByteBuffer buffer);
}
|
然后让FileReader 实现Iterable接口。每一个迭代器将会处理下一个4096字节的数据，并使用Decoder把它们解码成一个对象的List集合。注意，FileReader 接收文件(files)的list对象，这样是很好的，因为它可以遍历数据，并且不需要考虑聚合的问题。顺便说一下，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);
}
@Override
public Iterator<List<T>> iterator() {
return new Iterator<List<T>>() {
private List<T> entries;
private long chunkPos = 0;
private MappedByteBuffer buffer;
private FileChannel channel;
@Override
public 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);
}
chunkPos += 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);
}
}
@Override
public List<T> next() {
List<T> res = entries;
entries = null;
return res;
}
@Override
public void remove() {
throw new UnsupportedOperationException();
}
};
}
}
|
下一个任务就是写一个Decoder 。针对逗号分隔的任何文本格式，编写一个TextRowDecoder 类。接收的参数是每行字段的数量和一个字段分隔符，返回byte的二维数组。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;
}
@Override
public byte[][] decode(ByteBuffer buffer) {
int lineStartPos = buffer.position();
int limit = buffer.limit();
while (buffer.hasRemaining()) {
byte b = buffer.get();
if (b == LF) {
int lineEndPos = buffer.position();
if (buffer.limit() < lineEndPos + 1) {
buffer.position(lineStartPos).limit(lineEndPos);
} else {
buffer.position(lineStartPos).limit(lineEndPos + 1);
}
byte[][] entry = parseRow(buffer.duplicate());
if (entry != null) {
buffer.position(lineEndPos);
buffer.limit(limit);
lineStartPos = 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();
int limit = buffer.limit();
buffer.position(fieldStartPos).limit(fieldEndPos);
fields[fieldNumber] = parseField(buffer.duplicate(), fieldNumber, fieldEndPos - fieldStartPos - 1);
fieldNumber++;
buffer.position(fieldEndPos);
buffer.limit(limit);
fieldStartPos = 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;
}
}
|
这是文件被处理的过程。每一个List包含的元素都从一个单独的buffer中解码，每一个元素都是被TextRowDecoder定义的byte二维数组。
TextRowDecoder decoder = new TextRowDecoder(4, comma);
FileReader<byte[][]> reader = FileReader.create(decoder, file.listFiles());
for (List<byte[][]> chunk : reader) {
}
|
我们可以在这里打住，不过还有额外的需求。每一行都包含一个时间戳，每一批都必须分组，使用时间段来代替buffers，如按照天分组、或者按照小时分组。我还想要遍历每一批的数据，因此，第一反应就是，为FileReader创建一个Iterable包装器，实现它的行为。一个额外的细节，每一个元素必须通过实现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);
}
@Override
public 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;
@Override
public 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 (entryIt == null || !entryIt.hasNext()) {
if (entriesIt.hasNext()) {
entryIt = entriesIt.next().iterator();
return true;
} else {
return 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;
}
@Override
public List<T> next() {
return queue.poll();
}
@Override
public void remove() {
throw new UnsupportedOperationException();
}
};
}
}
|
最后的处理代码通过引入这个函数并无太大变动，只有一个干净的且紧密的循环，不必关心文件、缓冲区、时间周期的分组元素。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) {
for (TrueFxData entry : entries) {
}
}
|
