Java多线程读取大文本文件并批量插入MongoDB的实战代码

package ebs.core.filereader; import java.nio.file.Files; import java.nio.file.Paths; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Executors; import java.util.concurrent.ThreadPoolExecutor; import org.apache.commons.net.util.Charsets; /** * 大文本读取处理工具 * @author Xiedd * */ public class FileReader { private ThreadPoolExecutor fixedThreadPool = null; private String fileName;//文件名全路径 private Long split;//分组行数 private Long finished = 0L;//累计完成数 private FileReader(){} /** * @param fileName 文件名全路径 * @param split 分组行数，即一个线程处理多少行数据 * @param pool 线程池最大并发数 */ public FileReader(String fileName,long split,int pool){ fixedThreadPool = (ThreadPoolExecutor)Executors.newFixedThreadPool(pool); this.fileName = fileName; this.split = split; } public synchronized void finished(){ finished++; System.out.println("Finished:"+finished); } /** * 按GBK编码读取逐行读取文本数据，并按分组执行处理逻辑 * @param handler 自定义处理逻辑 * @throws Exception */ public void readLines(OperationHandler handler) throws Exception{ readLines(handler,"GBK"); } /** * 按行读取文本数据，并按分组执行处理逻辑 * @param handler 自定义处理逻辑 * @param encoding 编码 * @throws Exception */ public void readLines(OperationHandler handler,String encoding) throws Exception{ long start = System.currentTimeMillis(); List<String> tempList = new ArrayList<String>(); long from = 1; int times = 1; List<String> slist = Files.readAllLines(Paths.get(fileName), Charsets.toCharset(encoding)); long totalLine = slist.size();//ss.count();//总行数 System.out.println("总行数："+totalLine); //ss.close(); long groupNum = 0;//分多少组 if(totalLine%split==0){ groupNum = totalLine/split; }else{ groupNum = (totalLine/split) + 1; } for(int i=0;i<slist.size();i++){ String temp = slist.get(i); if(from<=split){ tempList.add(temp); if(from==split){ fixedThreadPool.execute(new OperationThread(tempList,times,handler,this)); tempList = new ArrayList<String>(); from = 0; times++; }else if((from<split)&&(times==groupNum)){ if(from==totalLine-(times-1)*split){ fixedThreadPool.execute(new OperationThread(tempList,times,handler,this)); tempList = new ArrayList<String>(); } } } from++; } slist.clear(); slist = null; tempList = null; long taskCount = fixedThreadPool.getTaskCount(); System.out.println("线程池内共有："+taskCount+" 个线程"); //检测全部线程是否结束 checkFinished(taskCount,start); } private void checkFinished(long threadCount,long startTime){ while(true){//判断是否线程池全部完成 if(finished==threadCount){ long end = System.currentTimeMillis(); System.out.println("全部结束，耗时："+(end-startTime)); //fixedThreadPool.awaitTermination(3000, TimeUnit.MILLISECONDS); fixedThreadPool.shutdownNow(); finished = 0L; break; } fixedThreadPool.getCompletedTaskCount(); } } }