一.前言

本文是对大数据文本文件读取（按行读取）的优化，目前常规的方案（限于JDK）有三种，第一种LineNumberReader，第二种RandomAccessFile，第三种是内存映射文件（详见http://sgq0085.iteye.com/blog/1318622）在RandomAccessFile基础上调用getChannel().map(...)。

1.LineNumberReader

按行读取，只能从第一行向后遍历，到需要读取的行时开始读入，直到完成；在我的测试用例中，读取1000W行数据每次5万行，用时93秒，效率实测比RandomAccessFile要高，但读取一亿跳数据时效率太低了（因为每次都要从头遍历），因为测试时超过1个小时，放弃测试；

2.RandomAccessFile

实际不适用于这种大数据读取，RandomAccessFile是为了磁盘文件的随机访问，所以效率很低，1000w行测试时用时140秒，一亿行数据测试用时1438秒但由于可以通过getFilePointer方法记录位置，并通过seek方法指定读取位置，所以从理论上比较适用这种大数据按行读取的场景；

RandomAccessFile只能按照8859_1这种方法读取，所以需要对内容重新编码，方法如下

Java代码

1. new String(pin.getBytes("8859_1"), "")

3.内存映射文件

由于每行数据大小不同，内存映射文件在这种情况下不适用，其他情况请参考我的博客（详见http://sgq0085.iteye.com/blog/1318622）

二.解决方案

如果在RandomAccessFile基础上，整合内部缓冲区，效率会有提高，测试过程中1000w行数据用时1秒，1亿行数据用时103（比1438秒快了13倍左右）

BufferedRandomAccessFile

网上已经有实现，代码如下：

Java代码

1. package com.gqshao.file.io;
3. import java.io.File;
4. import java.io.FileNotFoundException;
5. import java.io.IOException;
6. import java.io.RandomAccessFile;
7. import java.util.Arrays;
9. public class BufferedRandomAccessFile extends RandomAccessFile {
10. static final int LogBuffSz_ = 16; // 64K buffer
11. public static final int BuffSz_ = (1 << LogBuffSz_);
12. static final long BuffMask_ = ~(((long) BuffSz_) - 1L);
14. private String path_;
16. /*
17. * This implementation is based on the buffer implementation in Modula-3's
18. * "Rd", "Wr", "RdClass", and "WrClass" interfaces.
19. */
20. private boolean dirty_; // true iff unflushed bytes exist
21. private boolean syncNeeded_; // dirty_ can be cleared by e.g. seek, so track sync separately
22. private long curr_; // current position in file
23. private long lo_, hi_; // bounds on characters in "buff"
24. private byte[] buff_; // local buffer
25. private long maxHi_; // this.lo + this.buff.length
26. private boolean hitEOF_; // buffer contains last file block?
27. private long diskPos_; // disk position
29. public BufferedRandomAccessFile(File file, String mode) throws IOException {
30. this(file, mode, 0);
31. }
33. public BufferedRandomAccessFile(File file, String mode, int size) throws IOException {
34. super(file, mode);
35. path_ = file.getAbsolutePath();
36. this.init(size);
37. }
39. /**
40. * Open a new <code>BufferedRandomAccessFile</code> on the file named
41. * <code>name</code> in mode <code>mode</code>, which should be "r" for
42. * reading only, or "rw" for reading and writing.
43. */
44. public BufferedRandomAccessFile(String name, String mode) throws IOException {
45. this(name, mode, 0);
46. }
48. public BufferedRandomAccessFile(String name, String mode, int size) throws FileNotFoundException {
49. super(name, mode);
50. path_ = name;
51. this.init(size);
52. }
54. private void init(int size) {
55. this.dirty_ = false;
56. this.lo_ = this.curr_ = this.hi_ = 0;
57. this.buff_ = (size > BuffSz_) ? new byte[size] : new byte[BuffSz_];
58. this.maxHi_ = (long) BuffSz_;
59. this.hitEOF_ = false;
60. this.diskPos_ = 0L;
61. }
63. public String getPath() {
64. return path_;
65. }
67. public void sync() throws IOException {
68. if (syncNeeded_) {
69. flush();
70. getChannel().force(true);
71. syncNeeded_ = false;
72. }
73. }
75. // public boolean isEOF() throws IOException
76. // {
77. // assert getFilePointer() <= length();
78. // return getFilePointer() == length();
79. // }
81. public void close() throws IOException {
82. this.flush();
83. this.buff_ = null;
84. super.close();
85. }
87. /**
88. * Flush any bytes in the file's buffer that have not yet been written to
89. * disk. If the file was created read-only, this method is a no-op.
90. */
91. public void flush() throws IOException {
92. this.flushBuffer();
93. }
95. /* Flush any dirty bytes in the buffer to disk. */
96. private void flushBuffer() throws IOException {
97. if (this.dirty_) {
98. if (this.diskPos_ != this.lo_)
99. super.seek(this.lo_);
100. int len = (int) (this.curr_ - this.lo_);
101. super.write(this.buff_, 0, len);
102. this.diskPos_ = this.curr_;
103. this.dirty_ = false;
104. }
105. }
107. /*
108. * Read at most "this.buff.length" bytes into "this.buff", returning the
109. * number of bytes read. If the return result is less than
110. * "this.buff.length", then EOF was read.
111. */
112. private int fillBuffer() throws IOException {
113. int cnt = 0;
114. int rem = this.buff_.length;
115. while (rem > 0) {
116. int n = super.read(this.buff_, cnt, rem);
117. if (n < 0)
118. break;
119. cnt += n;
120. rem -= n;
121. }
122. if ((cnt < 0) && (this.hitEOF_ = (cnt < this.buff_.length))) {
123. // make sure buffer that wasn't read is initialized with -1
124. Arrays.fill(this.buff_, cnt, this.buff_.length, (byte) 0xff);
125. }
126. this.diskPos_ += cnt;
127. return cnt;
128. }
130. /*
131. * This method positions <code>this.curr</code> at position <code>pos</code>.
132. * If <code>pos</code> does not fall in the current buffer, it flushes the
133. * current buffer and loads the correct one.<p>
134. *
135. * On exit from this routine <code>this.curr == this.hi</code> iff <code>pos</code>
136. * is at or past the end-of-file, which can only happen if the file was
137. * opened in read-only mode.
138. */
139. public void seek(long pos) throws IOException {
140. if (pos >= this.hi_ || pos < this.lo_) {
141. // seeking outside of current buffer -- flush and read
142. this.flushBuffer();
143. this.lo_ = pos & BuffMask_; // start at BuffSz boundary
144. this.maxHi_ = this.lo_ + (long) this.buff_.length;
145. if (this.diskPos_ != this.lo_) {
146. super.seek(this.lo_);
147. this.diskPos_ = this.lo_;
148. }
149. int n = this.fillBuffer();
150. this.hi_ = this.lo_ + (long) n;
151. } else {
152. // seeking inside current buffer -- no read required
153. if (pos < this.curr_) {
154. // if seeking backwards, we must flush to maintain V4
155. this.flushBuffer();
156. }
157. }
158. this.curr_ = pos;
159. }
161. public long getFilePointer() {
162. return this.curr_;
163. }
165. public long length() throws IOException {
166. // max accounts for the case where we have written past the old file length, but not yet flushed our buffer
167. return Math.max(this.curr_, super.length());
168. }
170. public int read() throws IOException {
171. if (this.curr_ >= this.hi_) {
172. // test for EOF
173. // if (this.hi < this.maxHi) return -1;
174. if (this.hitEOF_)
175. return -1;
177. // slow path -- read another buffer
178. this.seek(this.curr_);
179. if (this.curr_ == this.hi_)
180. return -1;
181. }
182. byte res = this.buff_[(int) (this.curr_ - this.lo_)];
183. this.curr_++;
184. return ((int) res) & 0xFF; // convert byte -> int
185. }
187. public int read(byte[] b) throws IOException {
188. return this.read(b, 0, b.length);
189. }
191. public int read(byte[] b, int off, int len) throws IOException {
192. if (this.curr_ >= this.hi_) {
193. // test for EOF
194. // if (this.hi < this.maxHi) return -1;
195. if (this.hitEOF_)
196. return -1;
198. // slow path -- read another buffer
199. this.seek(this.curr_);
200. if (this.curr_ == this.hi_)
201. return -1;
202. }
203. len = Math.min(len, (int) (this.hi_ - this.curr_));
204. int buffOff = (int) (this.curr_ - this.lo_);
205. System.arraycopy(this.buff_, buffOff, b, off, len);
206. this.curr_ += len;
207. return len;
208. }
210. public void write(int b) throws IOException {
211. if (this.curr_ >= this.hi_) {
212. if (this.hitEOF_ && this.hi_ < this.maxHi_) {
213. // at EOF -- bump "hi"
214. this.hi_++;
215. } else {
216. // slow path -- write current buffer; read next one
217. this.seek(this.curr_);
218. if (this.curr_ == this.hi_) {
219. // appending to EOF -- bump "hi"
220. this.hi_++;
221. }
222. }
223. }
224. this.buff_[(int) (this.curr_ - this.lo_)] = (byte) b;
225. this.curr_++;
226. this.dirty_ = true;
227. syncNeeded_ = true;
228. }
230. public void write(byte[] b) throws IOException {
231. this.write(b, 0, b.length);
232. }
234. public void write(byte[] b, int off, int len) throws IOException {
235. while (len > 0) {
236. int n = this.writeAtMost(b, off, len);
237. off += n;
238. len -= n;
239. this.dirty_ = true;
240. syncNeeded_ = true;
241. }
242. }
244. /*
245. * Write at most "len" bytes to "b" starting at position "off", and return
246. * the number of bytes written.
247. */
248. private int writeAtMost(byte[] b, int off, int len) throws IOException {
249. if (this.curr_ >= this.hi_) {
250. if (this.hitEOF_ && this.hi_ < this.maxHi_) {
251. // at EOF -- bump "hi"
252. this.hi_ = this.maxHi_;
253. } else {
254. // slow path -- write current buffer; read next one
255. this.seek(this.curr_);
256. if (this.curr_ == this.hi_) {
257. // appending to EOF -- bump "hi"
258. this.hi_ = this.maxHi_;
259. }
260. }
261. }
262. len = Math.min(len, (int) (this.hi_ - this.curr_));
263. int buffOff = (int) (this.curr_ - this.lo_);
264. System.arraycopy(b, off, this.buff_, buffOff, len);
265. this.curr_ += len;
266. return len;
267. }
268. }

三.测试

1.FileUtil

用于封装三种方案（LineNumberReader、RandomAccessFile、BufferedRandomAccessFile）的文件读取

Java代码

1. package com.gqshao.file.util;
3. import com.google.common.collect.Lists;
4. import com.google.common.collect.Maps;
5. import com.gqshao.file.io.BufferedRandomAccessFile;
6. import org.apache.commons.io.IOUtils;
7. import org.apache.commons.lang3.StringUtils;
9. import java.io.*;
10. import java.util.List;
11. import java.util.Map;
13. public class FileUtil {
15. /**
16. * 通过BufferedRandomAccessFile读取文件,推荐
17. *
18. * @param file 源文件
19. * @param encoding 文件编码
20. * @param pos 偏移量
21. * @param num 读取量
22. * @return pins文件内容，pos当前偏移量
23. */
24. public static Map<String, Object> BufferedRandomAccessFileReadLine(File file, String encoding, long pos, int num) {
25. Map<String, Object> res = Maps.newHashMap();
26. List<String> pins = Lists.newArrayList();
27. res.put("pins", pins);
28. BufferedRandomAccessFile reader = null;
29. try {
30. reader = new BufferedRandomAccessFile(file, "r");
31. reader.seek(pos);
32. for (int i = 0; i < num; i++) {
33. String pin = reader.readLine();
34. if (StringUtils.isBlank(pin)) {
35. break;
36. }
37. pins.add(new String(pin.getBytes("8859_1"), encoding));
38. }
39. res.put("pos", reader.getFilePointer());
40. } catch (Exception e) {
41. e.printStackTrace();
42. } finally {
43. IOUtils.closeQuietly(reader);
44. }
45. return res;
46. }
48. /**
49. * 通过RandomAccessFile读取文件，能出来大数据文件，效率低
50. *
51. * @param file 源文件
52. * @param encoding 文件编码
53. * @param pos 偏移量
54. * @param num 读取量
55. * @return pins文件内容，pos当前偏移量
56. */
57. public static Map<String, Object> readLine(File file, String encoding, long pos, int num) {
58. Map<String, Object> res = Maps.newHashMap();
59. List<String> pins = Lists.newArrayList();
60. res.put("pins", pins);
61. RandomAccessFile reader = null;
62. try {
63. reader = new RandomAccessFile(file, "r");
64. reader.seek(pos);
65. for (int i = 0; i < num; i++) {
66. String pin = reader.readLine();
67. if (StringUtils.isBlank(pin)) {
68. break;
69. }
70. pins.add(new String(pin.getBytes("8859_1"), encoding));
71. }
72. res.put("pos", reader.getFilePointer());
73. } catch (Exception e) {
74. e.printStackTrace();
75. } finally {
76. IOUtils.closeQuietly(reader);
77. }
78. return res;
79. }
81. /**
82. * 使用LineNumberReader读取文件，1000w行比RandomAccessFile效率高，无法处理1亿条数据
83. *
84. * @param file 源文件
85. * @param encoding 文件编码
86. * @param index 开始位置
87. * @param num 读取量
88. * @return pins文件内容
89. */
90. public static List<String> readLine(File file, String encoding, int index, int num) {
91. List<String> pins = Lists.newArrayList();
92. LineNumberReader reader = null;
93. try {
94. reader = new LineNumberReader(new InputStreamReader(new FileInputStream(file), encoding));
95. int lines = 0;
96. while (true) {
97. String pin = reader.readLine();
98. if (StringUtils.isBlank(pin)) {
99. break;
100. }
101. if (lines >= index) {
102. if (StringUtils.isNotBlank(pin)) {
103. pins.add(pin);
104. }
105. }
106. if (num == pins.size()) {
107. break;
108. }
109. lines++;
110. }
111. } catch (Exception e) {
112. e.printStackTrace();
113. } finally {
114. IOUtils.closeQuietly(reader);
115. }
116. return pins;
117. }
120. }

2.RandomAccessFileTest

测试方法，涉及到的randomFile只是一个掺杂中文的文本文件，可以自己随便写一个

Java代码

1. package com.gqshao.file;
3. import com.gqshao.file.util.FileUtil;
4. import org.apache.commons.collections.CollectionUtils;
5. import org.apache.commons.collections.MapUtils;
6. import org.apache.commons.io.IOUtils;
7. import org.junit.Test;
8. import org.slf4j.Logger;
9. import org.slf4j.LoggerFactory;
11. import java.io.*;
12. import java.util.List;
13. import java.util.Map;
15. public class RandomAccessFileTest {
17. private static final Logger logger = LoggerFactory.getLogger(RandomAccessFileTest.class);
19. private static final String ENCODING = "UTF-8";
20. private static final int NUM = 50000;
22. private static File file = new File(ClassLoader.getSystemResource("").getPath() + File.separator + "test.txt");
23. private static File randomFile = new File(ClassLoader.getSystemResource("").getPath() + File.separator + "RandomFile.txt");
25. /**
26. * 生成1000w随机文本文件
27. */
28. @Test
29. public void makePin() {
30. String prefix = "_$#";
31. OutputStreamWriter out = null;
32. try {
33. out = new OutputStreamWriter(new FileOutputStream(file, true), ENCODING);
34. // 在1500w里随机1000w数据
35. for (int j = 0; j < 100000000; j++) {
36. out.write(prefix + (int) (130000000 * Math.random()) + "\n");
37. }
38. } catch (Exception e) {
39. e.printStackTrace();
40. } finally {
41. IOUtils.closeQuietly(out);
42. }
43. logger.info(file.getAbsolutePath());
44. }
46. /**
47. * 测试RandomAccessFile读取文件
48. */
49. @Test
50. public void testRandomAccessRead() {
51. long start = System.currentTimeMillis();
52. //
53. logger.info(String.valueOf(file.exists()));
54. long pos = 0L;
55. while (true) {
56. Map<String, Object> res = FileUtil.readLine(file, ENCODING, pos, NUM);
57. // 如果返回结果为空结束循环
58. if (MapUtils.isEmpty(res)) {
59. break;
60. }
61. Object po = res.get("pins");
62. List<String> pins = (List<String>) res.get("pins");
63. if (CollectionUtils.isNotEmpty(pins)) {
64. // logger.info(Arrays.toString(pins.toArray()));
65. if (pins.size() < NUM) {
66. break;
67. }
68. } else {
69. break;
70. }
71. pos = (Long) res.get("pos");
72. }
73. logger.info(((System.currentTimeMillis() - start) / 1000) + "");
74. }
76. /**
77. * 测试RandomAccessFile读取文件
78. */
79. @Test
80. public void testBufferedRandomAccessRead() {
81. long start = System.currentTimeMillis();
82. //
83. logger.info(String.valueOf(file.exists()));
84. long pos = 0L;
85. while (true) {
86. Map<String, Object> res = FileUtil.BufferedRandomAccessFileReadLine(file, ENCODING, pos, NUM);
87. // 如果返回结果为空结束循环
88. if (MapUtils.isEmpty(res)) {
89. break;
90. }
91. List<String> pins = (List<String>) res.get("pins");
92. if (CollectionUtils.isNotEmpty(pins)) {
93. // logger.info(Arrays.toString(pins.toArray()));
94. if (pins.size() < NUM) {
95. break;
96. }
97. } else {
98. break;
99. }
100. pos = (Long) res.get("pos");
101. }
102. logger.info(((System.currentTimeMillis() - start) / 1000) + "");
103. }
105. /**
106. * 测试普通读取文件
107. */
108. @Test
109. public void testCommonRead() {
110. long start = System.currentTimeMillis();
111. logger.info(String.valueOf(randomFile.exists()));
112. int index = 0;
113. while (true) {
114. List<String> pins = FileUtil.readLine(file, ENCODING, index, NUM);
115. if (CollectionUtils.isNotEmpty(pins)) {
116. // logger.info(Arrays.toString(pins.toArray()));
117. if (pins.size() < NUM) {
118. break;
119. }
120. } else {
121. break;
122. }
123. index += NUM;
124. }
125. logger.info(((System.currentTimeMillis() - start) / 1000) + "");
126. }
127. }