中文字幕av专区_日韩电影在线播放_精品国产精品久久一区免费式_av在线免费观看网站

溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

java使用多線程讀取超大文件

發布時間:2020-10-11 15:16:35 來源:腳本之家 閱讀:204 作者:Okey 欄目:編程語言

接上次寫的“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來享受風一般的速度吧!

以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持億速云。

向AI問一下細節

免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI

漾濞| 星座| 江北区| 肥东县| 上栗县| 惠来县| 定远县| 辽源市| 灵山县| 洪泽县| 洛浦县| 伊吾县| 进贤县| 林周县| 固阳县| 岳阳县| 江油市| 彩票| 海伦市| 壤塘县| 奇台县| 勃利县| 广西| 桦甸市| 孝义市| 桐柏县| 抚州市| 色达县| 南漳县| 保山市| 洪江市| 铁岭县| 台中县| 河源市| 彰武县| 韩城市| 静宁县| 商水县| 哈密市| 崇阳县| 定远县|