日常网络和数据库文件输入输出处理是很耗时间和消耗处理器时间的，所以I/O操作被普遍认为是昂贵的操作。 这里我们假设有一份非常大的文件，比如1G， 我们不可能拿任意一个InputStream 实现类去直接用那文件类File去封装传入构造方法直接处理，操作系统是不会让一个正在执行“非常耗时”的程序去占用大量时间去处理一个I/O 操作。所以我们必须拆分这个大文件为数个小的文件去依此处理。主体思想是先拆分成小字节块的文件在由这些小字节块文件合并成原来文件；这有点类似与网络传输一个大文件的过程，把文件拆除小的数据包然后再由数千甚至上万不等地路径传输到不同的终端。
主要核心：
- 按某个数量值去拆分大文件成小文件
- 小文件的读写都由一个特定的
BufferSize（字节容器大小）来决定输入和输出流的字节负荷
- 小文件的读写都由一个特定的
-
合并 的思路也类似
- 按特定的
BufferSize（字节容器大小）来决定读写可承受的负荷去处理多个拆分的小文件
- 按特定的
下方带图阐述了主要思路
注意 ： 本教程未用到同步和并发处理，仅是为了阐述I/O大文件拆分和合并的主体思想，若要用并发处理，下列代码实现需做改变。
FileSpliterAndMerger.png
FileSpliter.java
- 成员变量
- String inputPathName;- String outputPathNamePrefix;- int maxBufferSize;- final int ONE_BYTE = 1024;
- 成员方法
+ FileSpliter(String inputPathName, String outputPathNamePrefix, int maxBufferSize)+ void splitStart(int numOfSplits)+ void split(int numOfSplits) throws Exception+ static void readWrite(RandomAccessFile raf, BufferedOutputStream bw, long numBytes) throws IOException
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
/**
* FileSpliter: 用于拆分一个大容量文件，把其分成 numOfSplits 小块文件大小字节
* 处理，这个减缓IOStream的吞吐量。这个类可以日后加多线程处理或用
* 并行库来用不同的core或者thread去运算和处理每个文件。
* @author kaili
*
*/
public class FileSpliter {
private String inputPathName; // 输入文件路径
private String outputPathNamePrefix; // 输出文件路径
private int maxBufferSize; // 每个块的数据处理容器大小: maxBufferSize * ONE_BYTE = maxBufferSize kb
private final int ONE_BYTE = 1024;
/**
* 初始化下列参数列表成员变量
* @param inputPathName
* @param outputPathNamePrefix
* @param maxBufferSize
*/
public FileSpliter(String inputPathName, String outputPathNamePrefix,
int maxBufferSize) {
this.inputPathName = inputPathName;
this.outputPathNamePrefix = outputPathNamePrefix;
this.maxBufferSize = maxBufferSize;
}
/***
* 启动方法，间接调用拆分方法split
* @param numOfSplits
*/
public void splitStart(int numOfSplits) {
try {
split(numOfSplits);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 辅助方法
* @param numOfSplits 拆分文件块数量
* @throws Exception FileNotFoundException, IOException
*/
public void split(int numOfSplits) throws Exception {
File inputFile = new File(inputPathName);
RandomAccessFile raf = new RandomAccessFile(inputFile, "r");
long numSplits = numOfSplits;
// 记录随机访问输入流的大小
long sourceSize = raf.length();
// 算出每个拆分块的大小
long bytesPerSplit = sourceSize/numSplits ;
// 算出没被拆分最大值整除的剩余字节
long remainingBytes = sourceSize % numSplits;
// 最大的字节数组容器大小，便于数据的读写操作
int maxReadBufferSize = maxBufferSize * ONE_BYTE; // max size bytes taken in buffer
// 遍历所有的拆分块
for(int destIx=0; destIx < numSplits; destIx++) {
// 自制合并拆分文件的大小，用前缀加拆分下标destIX
String filePath = outputPathNamePrefix + "split."+ (destIx + 1);
// 用缓冲输出流来拆分字节文件输出到文件夹
BufferedOutputStream bw = new BufferedOutputStream(new FileOutputStream(filePath));
// 若每个拆分块字节大小比最大容器字节大小还大，我们
// 遍历处理这个文件字节块
if(bytesPerSplit > maxReadBufferSize) {
// 算出字节块能容多少个最大容器字节
long numReads = bytesPerSplit/maxReadBufferSize;
// 算出上个算式剩下不能被最大容器大小的字节数量
long numRemainingRead = bytesPerSplit % maxReadBufferSize;
// 把numReads * maxReadBufferSize 用遍历
// 形式去调用readWrite方法 把字节块拆分为
// 以maxReadBufferSize 容器大小的字节量依依读写
for(int i=0; i<numReads; i++) {
readWrite(raf, bw, maxReadBufferSize);
}
// 这里读写原理同上， 只不过读写的不是maxReadBufferSize
// 的大小，而是剩余的字节量
if(numRemainingRead > 0) {
readWrite(raf, bw, numRemainingRead);
}
}else {
// 若拆分字节块的字节量没有maxReadBufferSize大
// 直接调用readWrite处理
readWrite(raf, bw, bytesPerSplit);
}
bw.flush(); // 记得冲掉bw数据“管道”里的存储数据，以免有遗留
bw.close(); // 关闭流
}
// 这个就是拆分的字节块后剩余的字节，若此大于0， 字节用readWrite处理
if(remainingBytes > 0) {
BufferedOutputStream bw = new BufferedOutputStream(new FileOutputStream("split."+(numSplits+1)));
readWrite(raf, bw, remainingBytes);
bw.flush();
bw.close();
}
raf.close();
}
/**
* 辅助方法，读写功能，这个基本做了基本的数据处理工作； 方法要求下列参数。
* @param fis input file stream
* @param fos output file stream
* @param numBytes number of bytes to process in each buffer
* @return
* @throws IOException
*/
public static void readWrite(RandomAccessFile raf, BufferedOutputStream bw, long numBytes) throws IOException {
// 处理字节容器数组，用于传入输入流：这里是随机访问输入流
byte[] buf = new byte[(int) numBytes];
int val = raf.read(buf);
// 若返回字节值不等于-1， 写字节数据到缓冲输出流的指定文件
if(val != -1) {
bw.write(buf);
}
}
}
FileMerger.java
-
成员变量
- FileInputStream fis;- FileOutputStream fos;- List<File> list;- String inputMergerFilePath;- String outputMergerFilePath;- int maxBufferChunkSize;- int ONE_BYTE = 1024;
-
成员方法
+ FileMerger(String inputMergerPath, String outputMergerPath, int maxBufferSize)+ void startMerging()+ void merge()+ void processFilesInList(File ofile)+ void sortList(List<File> fileLists)+ long readWrite(FileInputStream fis, FileOutputStream fos, long numBytes) throws IOException
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
/**
* FileMerger: 用于合并拆分后的文件夹中的文件
* @author kaili
*
*/
public class FileMerger {
private FileInputStream fis;
private FileOutputStream fos;
private List<File> list;
private String inputMergerFilePath;
private String outputMergerFilePath;
private int maxBufferChunkSize;
private int ONE_BYTE = 1024;
/***
* 初始化 输入文件路径 和 输出文件路径 和 最大 容器 大小 （用于存储每次读写的容器字节数组）
* @param inputMergerPath input path
* @param outputMergerPath output path
* @param maxBufferSize the maximum size of each split chunk or size of buffer
*/
public FileMerger(String inputMergerPath, String outputMergerPath, int maxBufferSize) {
this.inputMergerFilePath = inputMergerPath;
this.outputMergerFilePath = outputMergerPath;
this.maxBufferChunkSize = maxBufferSize * ONE_BYTE; // the maximum size of each split chunk
}
/**
* 开始合并
*/
public void startMerging() {
merge();
}
/**
* 辅助方法用于合并
*/
public void merge() {
// 合并（拆分文件)的输出文件对象
File ofile = new File(outputMergerFilePath);
// 输入文件类（本质必须是一个文件夹）
File inputFile = new File(inputMergerFilePath);
// 赋值文件列表
this.list = new ArrayList<File>();
if (inputFile.isDirectory()) { // 检查是不是文件夹
// 拿到文件夹的里面的文件数组
File[] listFiles = inputFile.listFiles();
// 加载所有拆分文件到列表集合里头
for (File inputf : listFiles) {
list.add(inputf);
}
// 注意： inputFile.listFiles() 传回的数组是无序的
// 所以拆分文件命名必须要遵守某种规则，便于取出文件的序列号
// 也为排序做好铺垫
// 排序好数组里的文件
sortList(list);
// 处理列表里的文件
processFilesInList(ofile);
} else {
System.err.println("合并不成功，处理文件必须要是文件夹.");
}
}
/**
* 处理文件方法，遍历所有文件数组里的文件
* 注意：在多线程或并发环境条件下，这个拆分处理
// 需要应相应情况而改变。
* @param ofile
*/
public void processFilesInList(File ofile) {
try {
// 输出流
fos = new FileOutputStream(ofile,true);
// 遍历文件
for (File file : list) {
fis = new FileInputStream(file);
// 这个变量用于检查拆分字节量是否等于原文件大小
long sumByte = 0;
// 如文件字节大小大于最大字节容器大小
if (file.length() > maxBufferChunkSize) {
// 算出从文件字节里可以拆分多少个以maxBufferChunkSize
// 为大小的单位
long numReads = file.length() / maxBufferChunkSize;
// 算出剩余的拆分后的字节大小
long remains = file.length() % maxBufferChunkSize;
// 读写numReads*maxBufferChunkSize大小的字节量
for (int i = 0; i < numReads; i++) {
sumByte += readWrite(fis, fos, maxBufferChunkSize);
}
// 若剩余字节大于零，直接读写
if (remains > 0) {
sumByte += readWrite(fis, fos, remains);
}
} else { // 文件大小小于最大容器数量，直接读写
sumByte += readWrite(fis, fos, file.length());
}
// 检查拆分字节量的总量是否等于文件字节量
assert(sumByte == file.length());
fis.close(); // 关闭输入流
fis = null; // 重置输入流
}
fos.close(); // 关闭输出流
fos = null; // 重置输出流
}catch (Exception exception){
exception.printStackTrace();
}
}
/**
* 按文件序列号排序，从小到大。文件名：拆分名.序号
*
* 合同：拆分文件的合并必须时有序的，所以文件命名要遵循一定规则
* @param fileLists 文件数组列表
*/
public void sortList(List<File> fileLists) {
Collections.sort(fileLists, new Comparator<File>() {
@Override
public int compare(File o1, File o2) {
String fileName1 = o1.getName();
String fileName2 = o2.getName();
int val1 = Integer.parseInt(fileName1.substring(fileName1.lastIndexOf('.') + 1, fileName1.length()));
int val2 = Integer.parseInt(fileName2.substring(fileName2.lastIndexOf('.') + 1, fileName2.length()));
if (val1 < val2) {
return -1;
} else if (val1 > val2) {
return 1;
} else {
return 0;
}
}
});
}
/**
* 读写赋值方法：要求下列参数。
* @param fis input file stream
* @param fos output file stream
* @param numBytes number of bytes to process in each buffer
* @return
* @throws IOException
*/
public long readWrite(FileInputStream fis, FileOutputStream fos, long numBytes) throws IOException {
byte[] buf = new byte[(int) numBytes];
int bytesRead = fis.read(buf, 0, (int)numBytes); // 3个变量的读操作会帮助fis得知处理字节数量
if(bytesRead != -1) {
fos.write(buf);
}
assert(bytesRead == numBytes);
fos.flush();
return bytesRead;
}
}
main 方法实现
private static final int MAX_BUFFER_SIZE = 8; // 8 kb
private static final int NUM_OF_SPLITS = 10; // 10 SPLITS
public static void main(String[] args) {
// -------------------------------Splitter---------------------------------//
// 注意： 需要改变路径来测试
String inputSplitPath = "---------xx路径---------";
String outputSplitPathPrefix = "----------------xx路径---------------";
//---------------------------------------------------------------//
System.out.println("start splitting...");
Instant start = Instant.now();
FileSpliter fs = new FileSpliter(inputSplitPath, outputSplitPathPrefix, MAX_BUFFER_SIZE);
fs.splitStart(NUM_OF_SPLITS);
Instant end = Instant.now();
System.out.println(Duration.between(start, end).getSeconds());
System.out.println("done spliting...");
// ---------------------------------Merger-------------------------------- //
System.out.println("start merging...");
// 注意： 需要改变路径来测试
String inputMergerFilePath = "----------------xx路径---------------";
String outputMergerFilePath = "----------------xx路径---------------";
start = Instant.now();
FileMerger fm1 = new FileMerger(inputMergerFilePath, outputMergerFilePath, MAX_BUFFER_SIZE);
fm1.startMerging();
end = Instant.now();
System.out.println(Duration.between(start, end).getSeconds());
System.out.println("done merging...");
}
