我最近要处理一套存储历史实时数据的大文件fx market data，我很快便意识到，使用传统的InputStream不能够将它们读取到内存，因为每一个文件都超过了4G。甚至编辑器都不能够打开这些文件。
在这种特殊情况下，我可以写一个简单的bash脚本将这些文件分成更小的文件块，然后再读取它。但是我不想这样做，因为二进制格式会使这个方法失效。
处理这个问题的方式通常就是使用内存映射文件递增地处理区域的数据。关于内存映射文件的一个好处就是它们不会使用虚拟内存和换页空间，因为它们是从磁盘上的文件返回来的数据。
很好，让我们来看一看这些文件和额外的一些数据。似乎它们使用逗号分隔的字段包含ASCII文本行。
格式：[currency-pair],[timestamp],[bid-price],[ask-price]
例子：EUR/USD,20120102 00:01:30.420,1.29451,1.2949
我可以为这种格式去写一个程序，但是，读取文件和解析文件是无关的概念。让我们退一步来想一个通用的设计，当在将来面临相似的问题时这个设计可以被重复利用。
这个问题可以归结为递增地解码一个已经在无限长的数组中被编码的记录，并且没有耗尽内存。实际上，以逗号分割的示例格式编码与通常的解决方案是不相关的。所以，很明显需要一个解码器来处理不同的格式。
再来看，知道整个文件处理完成，每一条记录都不能被解析并保存在内存中，所以我们需要一种方式来转移记录，在它们成为垃圾被回收之前可以被写到其他地方，例如磁盘或者网络。
迭代器是处理这个需求的很好的抽象，因为它们就像游标一样，可以正确的指向某个位置。每一次迭代都可以转发文件指针，并且可以让我们使用数据做其他的事情。
首先来写一个Decoder 接口，递增地把对象从MappedByteBuffer中解码，如果buffer中没有对象，则返回null。
|
1
2
3
|
public
interface Decoder<T> { public
T decode(ByteBuffer buffer);} |
然后让FileReader 实现Iterable接口。每一个迭代器将会处理下一个4096字节的数据，并使用Decoder把它们解码成一个对象的List集合。注意，FileReader 接收文件(files)的list对象，这样是很好的，因为它可以遍历数据，并且不需要考虑聚合的问题。顺便说一下，4096个字节块对于大文件来说是非常小的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
|
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); } // set next MappedByteBuffer chunk 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 可以被操作不同字符集的特定格式解码器重复利用。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
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) { // reached line feed so parse line int
lineEndPos = buffer.position(); // set positions for one row duplication 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) { // reset main buffer buffer.position(lineEndPos); buffer.limit(limit); // set start after LF 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(); // save limit int
limit = buffer.limit(); // set positions for one row duplication buffer.position(fieldStartPos).limit(fieldEndPos); fields[fieldNumber] = parseField(buffer.duplicate(), fieldNumber, fieldEndPos - fieldStartPos -
1); fieldNumber++; // reset main buffer buffer.position(fieldEndPos); buffer.limit(limit); // set start after LF 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二维数组。
|
1
2
3
4
5
|
TextRowDecoder decoder =
new TextRowDecoder(4, comma);FileReader<byte[][]> reader = FileReader.create(decoder, file.listFiles());for
(List<byte[][]> chunk : reader) { // do something with each chunk} |
我们可以在这里打住，不过还有额外的需求。每一行都包含一个时间戳，每一批都必须分组，使用时间段来代替buffers，如按照天分组、或者按照小时分组。我还想要遍历每一批的数据，因此，第一反应就是，为FileReader创建一个Iterable包装器，实现它的行为。一个额外的细节，每一个元素必须通过实现Timestamped接口（这里没有显示）提供时间戳到PeriodEntries。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
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 there are no rows left if
(entryIt == null
|| !entryIt.hasNext()) { // try get more rows if possible if
(entriesIt.hasNext()) { entryIt = entriesIt.next().iterator(); return
true; }
else { // no more rows 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也是足够的灵活管理任何时长的时间。
|
1
2
3
4
5
6
7
8
9
10
11
|
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 day for
(TrueFxData entry : entries) { // process each entry }} |
你也许意识到了，使用集合不可能解决这样的问题；选择迭代器是一个关键的设计决策，能够解析兆字节的数组，且不会消耗过多的空间。
原文链接： javacodegeeks 翻译： ImportNew - 踏雁寻花