接上次写的“java读取超大文件”。在读取超过10g的文件时会发现一次读一行的速度实在是不能接受，想到使用多线程+filechannel来做一个使用多线程版本。
基本思路如下：
1.计算出文件总大小
2.分段处理,计算出每个线程读取文件的开始与结束位置
(文件大小/线程数)*n,n是指第几个线程,这样能得到每个线程在读该文件的大概起始位置
使用"大概起始位置",作为读文件的开始偏移量(filechannel.position("大概起始位置")),来读取该文件,直到读到第一个换行符,记录下这个换行符的位置,作为该线程的准确起 始位置.同时它也是上一个线程的结束位置.最后一个线程的结束位置也直接设置为-1
3.启动线程,每个线程从开始位置读取到结束位置为止
代码如下:
读文件工具类
我要评论
import java.io.*;
import java.nio.bytebuffer;
import java.nio.channels.filechannel;
import java.util.observable;
/**
* created with intellij idea.
* user: okey
* date: 14-4-2
* time: 下午3:12
* 读取文件
*/
public class readfile extends observable {
private int bufsize = 1024;
// 换行符
private byte key = "\n".getbytes()[0];
// 当前行数
private long linenum = 0;
// 文件编码,默认为gb2312
private string encode = "gb2312";
// 具体业务逻辑监听器
private readerfilelistener readerlistener;
public void setencode(string encode) {
this.encode = encode;
}
public void setreaderlistener(readerfilelistener readerlistener) {
this.readerlistener = readerlistener;
}
/**
* 获取准确开始位置
* @param file
* @param position
* @return
* @throws exception
*/
public long getstartnum(file file, long position) throws exception {
long startnum = position;
filechannel fcin = new randomaccessfile(file, "r").getchannel();
fcin.position(position);
try {
int cache = 1024;
bytebuffer rbuffer = bytebuffer.allocate(cache);
// 每次读取的内容
byte[] bs = new byte[cache];
// 缓存
byte[] tempbs = new byte[0];
string line = "";
while (fcin.read(rbuffer) != -1) {
int rsize = rbuffer.position();
rbuffer.rewind();
rbuffer.get(bs);
rbuffer.clear();
byte[] newstrbyte = bs;
// 如果发现有上次未读完的缓存,则将它加到当前读取的内容前面
if (null != tempbs) {
int tl = tempbs.length;
newstrbyte = new byte[rsize + tl];
system.arraycopy(tempbs, 0, newstrbyte, 0, tl);
system.arraycopy(bs, 0, newstrbyte, tl, rsize);
}
// 获取开始位置之后的第一个换行符
int endindex = indexof(newstrbyte, 0);
if (endindex != -1) {
return startnum + endindex;
}
tempbs = substring(newstrbyte, 0, newstrbyte.length);
startnum += 1024;
}
} catch (exception e) {
e.printstacktrace();
} finally {
fcin.close();
}
return position;
}
/**
* 从设置的开始位置读取文件，一直到结束为止。如果 end设置为负数,刚读取到文件末尾
* @param fullpath
* @param start
* @param end
* @throws exception
*/
public void readfilebyline(string fullpath, long start, long end) throws exception {
file fin = new file(fullpath);
if (fin.exists()) {
filechannel fcin = new randomaccessfile(fin, "r").getchannel();
fcin.position(start);
try {
bytebuffer rbuffer = bytebuffer.allocate(bufsize);
// 每次读取的内容
byte[] bs = new byte[bufsize];
// 缓存
byte[] tempbs = new byte[0];
string line = "";
// 当前读取文件位置
long nowcur = start;
while (fcin.read(rbuffer) != -1) {
nowcur += bufsize;
int rsize = rbuffer.position();
rbuffer.rewind();
rbuffer.get(bs);
rbuffer.clear();
byte[] newstrbyte = bs;
// 如果发现有上次未读完的缓存,则将它加到当前读取的内容前面
if (null != tempbs) {
int tl = tempbs.length;
newstrbyte = new byte[rsize + tl];
system.arraycopy(tempbs, 0, newstrbyte, 0, tl);
system.arraycopy(bs, 0, newstrbyte, tl, rsize);
}
// 是否已经读到最后一位
boolean isend = false;
// 如果当前读取的位数已经比设置的结束位置大的时候，将读取的内容截取到设置的结束位置
if (end > 0 && nowcur > end) {
// 缓存长度 - 当前已经读取位数 - 最后位数
int l = newstrbyte.length - (int) (nowcur - end);
newstrbyte = substring(newstrbyte, 0, l);
isend = true;
}
int fromindex = 0;
int endindex = 0;
// 每次读一行内容，以 key（默认为\n） 作为结束符
while ((endindex = indexof(newstrbyte, fromindex)) != -1) {
byte[] bline = substring(newstrbyte, fromindex, endindex);
line = new string(bline, 0, bline.length, encode);
linenum++;
// 输出一行内容，处理方式由调用方提供
readerlistener.outline(line.trim(), linenum, false);
fromindex = endindex + 1;
}
// 将未读取完成的内容放到缓存中
tempbs = substring(newstrbyte, fromindex, newstrbyte.length);
if (isend) {
break;
}
}
// 将剩下的最后内容作为一行，输出，并指明这是最后一行
string linestr = new string(tempbs, 0, tempbs.length, encode);
readerlistener.outline(linestr.trim(), linenum, true);
} catch (exception e) {
e.printstacktrace();
} finally {
fcin.close();
}
} else {
throw new filenotfoundexception("没有找到文件：" + fullpath);
}
// 通知观察者,当前工作已经完成
setchanged();
notifyobservers(start+"-"+end);
}
/**
* 查找一个byte[]从指定位置之后的一个换行符位置
*
* @param src
* @param fromindex
* @return
* @throws exception
*/
private int indexof(byte[] src, int fromindex) throws exception {
for (int i = fromindex; i < src.length; i++) {
if (src[i] == key) {
return i;
}
}
return -1;
}
/**
* 从指定开始位置读取一个byte[]直到指定结束位置为止生成一个全新的byte[]
*
* @param src
* @param fromindex
* @param endindex
* @return
* @throws exception
*/
private byte[] substring(byte[] src, int fromindex, int endindex) throws exception {
int size = endindex - fromindex;
byte[] ret = new byte[size];
system.arraycopy(src, fromindex, ret, 0, size);
return ret;
}
}
读文件线程
/**
* created with intellij idea.
* user: okey
* date: 14-4-2
* time: 下午4:50
* to change this template use file | settings | file templates.
*/
public class readfilethread extends thread {
private readerfilelistener processpoidatalisteners;
private string filepath;
private long start;
private long end;
public readfilethread(readerfilelistener processpoidatalisteners,long start,long end,string file) {
this.setname(this.getname()+"-readfilethread");
this.start = start;
this.end = end;
this.filepath = file;
this.processpoidatalisteners = processpoidatalisteners;
}
@override
public void run() {
readfile readfile = new readfile();
readfile.setreaderlistener(processpoidatalisteners);
readfile.setencode(processpoidatalisteners.getencode());
// readfile.addobserver();
try {
readfile.readfilebyline(filepath, start, end + 1);
} catch (exception e) {
e.printstacktrace();
}
}
}
具体业务逻辑监听
/**
* created with okey
* user: okey
* date: 13-3-14
* time: 下午3:19
* nio逐行读数据回调方法
*/
public abstract class readerfilelistener {
// 一次读取行数，默认为500
private int readcolnum = 500;
private string encode;
private list<string> list = new arraylist<string>();
/**
* 设置一次读取行数
* @param readcolnum
*/
protected void setreadcolnum(int readcolnum) {
this.readcolnum = readcolnum;
}
public string getencode() {
return encode;
}
public void setencode(string encode) {
this.encode = encode;
}
/**
* 每读取到一行数据，添加到缓存中
* @param linestr 读取到的数据
* @param linenum 行号
* @param over 是否读取完成
* @throws exception
*/
public void outline(string linestr, long linenum, boolean over) throws exception {
if(null != linestr)
list.add(linestr);
if (!over && (linenum % readcolnum == 0)) {
output(list);
list.clear();
} else if (over) {
output(list);
list.clear();
}
}
/**
* 批量输出
*
* @param stringlist
* @throws exception
*/
public abstract void output(list<string> stringlist) throws exception;
}
线程调度
import java.io.file;
import java.io.fileinputstream;
import java.io.ioexception;
/**
* created with intellij idea.
* user: okey
* date: 14-4-1
* time: 下午6:03
* to change this template use file | settings | file templates.
*/
public class builddata {
public static void main(string[] args) throws exception {
file file = new file("e:\\1396341974289.csv");
fileinputstream fis = null;
try {
readfile readfile = new readfile();
fis = new fileinputstream(file);
int available = fis.available();
int maxthreadnum = 50;
// 线程粗略开始位置
int i = available / maxthreadnum;
for (int j = 0; j < maxthreadnum; j++) {
// 计算精确开始位置
long startnum = j == 0 ? 0 : readfile.getstartnum(file, i * j);
long endnum = j + 1 < maxthreadnum ? readfile.getstartnum(file, i * (j + 1)) : -2;
// 具体监听实现
processdatabypostgislisteners listeners = new processdatabypostgislisteners("gbk");
new readfilethread(listeners, startnum, endnum, file.getpath()).start();
}
} catch (ioexception e) {
e.printstacktrace();
} catch (exception e) {
e.printstacktrace();
}
}
}
现在就可以尽情的调整 maxthreadnum来享受风一般的速度吧!
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持移动技术网。
如对本文有疑问， 点击进行留言回复！！
网友评论