Java多线程实现文件快速切分详解编程语言

[Java]代码    

import org.apache.log4j.LogManager; 
import org.apache.log4j.Logger; 
import java.io.*; 
import java.util.*; 
import java.util.concurrent.*; 
public class FileSplitUtil { 
private final static Logger log = LogManager.getLogger(FileSplitUtil.class); 
private static final long originFileSize = 1024 * 1024 * 100;// 100M 
private static final int blockFileSize = 1024 * 1024 * 64;// 防止中文乱码，必须取2的N次方 
/** 
* CVS文件分隔符 
*/ 
private static final char cvsSeparator = '^'; 
public static  void  main(String args[]){ 
long start = System.currentTimeMillis(); 
try { 
String fileName = "D://csvtest//aa.csv"; 
File sourceFile = new File(fileName); 
if (sourceFile.length() >= originFileSize) { 
String cvsFileName = fileName.replaceAll("////", "/"); 
FileSplitUtil fileSplitUtil = new FileSplitUtil(); 
List<String> parts=fileSplitUtil.splitBySize(cvsFileName, blockFileSize); 
for(String part:parts){ 
System.out.println("partName is:"+part); 
} 
} 
System.out.println("总文件长度"+sourceFile.length()+",拆分文件耗时:" + (System.currentTimeMillis() - start) + "ms."); 
}catch (Exception e){ 
log.info(e.getStackTrace()); 
} 
} 
/** 
* 拆分文件 
* 
* @param fileName 待拆分的完整文件名 
* @param byteSize 按多少字节大小拆分 
* @return 拆分后的文件名列表 
*/ 
public List<String> splitBySize(String fileName, int byteSize) 
throws IOException, InterruptedException { 
List<String> parts = new ArrayList<String>(); 
File file = new File(fileName); 
int count = (int) Math.ceil(file.length() / (double) byteSize); 
int countLen = (count + "").length(); 
RandomAccessFile raf = new RandomAccessFile(fileName, "r"); 
long totalLen = raf.length(); 
CountDownLatch latch = new CountDownLatch(count); 
for (int i = 0; i < count; i++) { 
String partFileName = file.getPath() + "." 
+ leftPad((i + 1) + "", countLen, '0') + ".cvs"; 
int readSize=byteSize; 
long startPos=(long)i * byteSize; 
long nextPos=(long)(i+1) * byteSize; 
if(nextPos>totalLen){ 
readSize= (int) (totalLen-startPos); 
} 
new SplitRunnable(readSize, startPos, partFileName, file, latch).run(); 
parts.add(partFileName); 
} 
latch.await();//等待所有文件写完 
//由于切割时可能会导致行被切断，加工所有的的分割文件,合并行 
mergeRow(parts); 
return parts; 
} 
/** 
* 分割处理Runnable 
* 
* @author supeidong 
*/ 
private class SplitRunnable implements Runnable { 
int byteSize; 
String partFileName; 
File originFile; 
long startPos; 
CountDownLatch latch; 
public SplitRunnable(int byteSize, long startPos, String partFileName, 
File originFile, CountDownLatch latch) { 
this.startPos = startPos; 
this.byteSize = byteSize; 
this.partFileName = partFileName; 
this.originFile = originFile; 
this.latch = latch; 
} 
public void run() { 
RandomAccessFile rFile; 
OutputStream os; 
try { 
rFile = new RandomAccessFile(originFile, "r"); 
byte[] b = new byte[byteSize]; 
rFile.seek(startPos);// 移动指针到每“段”开头 
int s = rFile.read(b); 
os = new FileOutputStream(partFileName); 
os.write(b, 0, s); 
os.flush(); 
os.close(); 
latch.countDown(); 
} catch (IOException e) { 
log.error(e.getMessage()); 
latch.countDown(); 
} 
} 
} 
/** 
* 合并被切断的行 
* 
* @param parts 
*/ 
private void mergeRow(List<String> parts) { 
List<PartFile> partFiles = new ArrayList<PartFile>(); 
try { 
//组装被切分表对象 
for (int i=0;i<parts.size();i++) { 
String partFileName=parts.get(i); 
File splitFileTemp = new File(partFileName); 
if (splitFileTemp.exists()) { 
PartFile partFile = new PartFile(); 
BufferedReader reader=new BufferedReader(new InputStreamReader(new FileInputStream(splitFileTemp),"gbk")); 
String firstRow = reader.readLine(); 
String secondRow = reader.readLine(); 
String endRow = readLastLine(partFileName); 
partFile.setPartFileName(partFileName); 
partFile.setFirstRow(firstRow); 
partFile.setEndRow(endRow); 
if(i>=1){ 
String prePartFile=parts.get(i - 1); 
String preEndRow = readLastLine(prePartFile); 
partFile.setFirstIsFull(getCharCount(firstRow+preEndRow)>getCharCount(secondRow)); 
} 
partFiles.add(partFile); 
reader.close(); 
} 
} 
//进行需要合并的行的写入 
for (int i = 0; i < partFiles.size() - 1; i++) { 
PartFile partFile = partFiles.get(i); 
PartFile partFileNext = partFiles.get(i + 1); 
StringBuilder sb = new StringBuilder(); 
if (partFileNext.getFirstIsFull()) { 
sb.append("/r/n"); 
sb.append(partFileNext.getFirstRow()); 
} else { 
sb.append(partFileNext.getFirstRow()); 
} 
writeLastLine(partFile.getPartFileName(),sb.toString()); 
} 
} catch (Exception e) { 
log.error(e.getMessage()); 
} 
} 
/** 
* 得到某个字符出现的次数 
* @param s 
* @return 
*/ 
private int getCharCount(String s) { 
int count = 0; 
for (int i = 0; i < s.length(); i++) { 
if (s.charAt(i) == cvsSeparator) { 
count++; 
} 
} 
return count; 
} 
/** 
* 采用BufferedInputStream方式读取文件行数 
* 
* @param filename 
* @return 
*/ 
public int getFileRow(String filename) throws IOException { 
InputStream is = new BufferedInputStream(new FileInputStream(filename)); 
byte[] c = new byte[1024]; 
int count = 0; 
int readChars = 0; 
while ((readChars = is.read(c)) != -1) { 
for (int i = 0; i < readChars; ++i) { 
if (c[i] == '/n') 
++count; 
} 
} 
is.close(); 
return count; 
} 
/** 
* 读取最后一行数据 
* @param filename 
* @return 
* @throws IOException 
*/ 
private String readLastLine(String filename) throws IOException { 
// 使用RandomAccessFile , 从后找最后一行数据 
RandomAccessFile raf = new RandomAccessFile(filename, "r"); 
long len = raf.length(); 
String lastLine = ""; 
if(len!=0L) { 
long pos = len - 1; 
while (pos > 0) { 
pos--; 
raf.seek(pos); 
if (raf.readByte() == '/n') { 
lastLine = raf.readLine(); 
lastLine=new String(lastLine.getBytes("8859_1"), "gbk"); 
break; 
} 
} 
} 
raf.close(); 
return lastLine; 
} 
/** 
* 修改最后一行数据 
* @param fileName 
* @param lastString 
* @return 
* @throws IOException 
*/ 
private void writeLastLine(String fileName,String lastString){ 
try { 
// 打开一个随机访问文件流，按读写方式 
RandomAccessFile randomFile = new RandomAccessFile(fileName, "rw"); 
// 文件长度，字节数 
long fileLength = randomFile.length(); 
//将写文件指针移到文件尾。 
randomFile.seek(fileLength); 
//此处必须加gbk，否则会出现写入乱码 
randomFile.write(lastString.getBytes("gbk")); 
randomFile.close(); 
} catch (IOException e) { 
log.error(e.getMessage()); 
} 
} 
/** 
* 左填充 
* 
* @param str 
* @param length 
* @param ch 
* @return 
*/ 
public static String leftPad(String str, int length, char ch) { 
if (str.length() >= length) { 
return str; 
} 
char[] chs = new char[length]; 
Arrays.fill(chs, ch); 
char[] src = str.toCharArray(); 
System.arraycopy(src, 0, chs, length - src.length, src.length); 
return new String(chs); 
} 
/** 
* 合并文件行内部类 
*/ 
class PartFile { 
private String partFileName; 
private String firstRow; 
private String endRow; 
private boolean firstIsFull; 
public String getPartFileName() { 
return partFileName; 
} 
public void setPartFileName(String partFileName) { 
this.partFileName = partFileName; 
} 
public String getFirstRow() { 
return firstRow; 
} 
public void setFirstRow(String firstRow) { 
this.firstRow = firstRow; 
} 
public String getEndRow() { 
return endRow; 
} 
public void setEndRow(String endRow) { 
this.endRow = endRow; 
} 
public boolean getFirstIsFull() { 
return firstIsFull; 
} 
public void setFirstIsFull(boolean firstIsFull) { 
this.firstIsFull = firstIsFull; 
} 
} 
}