Java 分割功能实现

    在项目针对比较大的文件需要文件的分割功能,特意写了一个具有分割功能代码,以供大家拍砖之用,目的进行沟通交流,如有好的建议和联系本人沟通交流谢谢!

package easyway.tbs.file.transport.core.split;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.Serializable;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * 文件读写访问器
 * 
 * @author longgangbai
 *
 */
public class DataFileAccessor implements Serializable {
	 private final static Log logger = LogFactory.getLog(DataFileAccessor.class);
     
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	/**
	 * 只读模式
	 */
	public static final String RANDOMACESSFILE_READ_MODE="r";
	/**
	 * 只写模式
	 */
	public static final String RANDOMACESSFILE_WRITE_MODE="w";
	/**
	 * 读写模式
	 */
	public static final String RANDOMACESSFILE_WRITE_READ_MODE="rw";
	
	/**
	 * 随机读取的缓存文件的长度
	 */
	public static final int RANDOMACESSFILE_BUFFER_LENGTH=1024*1000;
				
				
	/**
	 * 文件读取的对象操作对象
	 */
	private RandomAccessFile fileAcessor;
	/**
	 * 在文件的路径
	 */
    private String filePath;
    
    /**
     * 
     * @throws IOException
     */
	public DataFileAccessor() throws IOException {
		this("", RANDOMACESSFILE_WRITE_READ_MODE);
	}

	/**
	 * 文件读取访问其
	 * @param sName 
	 * @param nPos
	  * @param nPos
	 * @throws IOException
	 */
	public DataFileAccessor(String filePath, String mode) throws IOException {
		this.filePath=filePath;
		//支持读写操作
		if(logger.isDebugEnabled()){
			logger.info("filePath="+filePath+"======mode===="+mode);
		}
		fileAcessor = new RandomAccessFile(filePath, mode);
	}

	/**
	 * 写文件的片段
	 * @param buffer 
	 *             临时存储的数据的数组
	 * @param nStart
	 * @param nLen
	 * @return
	 */
	public synchronized int write(byte[] buffer, int nStart, int nLen) {
		if(logger.isDebugEnabled()){
			logger.info("nStart="+nStart+"  nLen="+nLen);
		}
		int n = -1;
		try {
			//移动文件指针
			fileAcessor.seek(nStart);
			fileAcessor.write(buffer, nStart, nLen);
			n = nLen;
		} catch (IOException e) {
			logger.error("write file "+filePath +" nStart="+nStart+"   nLen="+nLen,e);
			
		}
		return n;
	}
	
	/**
	 * 写文件的片段
	 * @param buffer 
	 *             临时存储的数据的数组
	 * @param nStart
	 * @param nLen
	 * @return
	 */
	public synchronized int readFile(byte[] buffer, int offset, int nLen) {
		try {
			fileAcessor.seek(offset);
			long length = nLen + 1;// 文件段的长度
			byte[] b = new byte[RANDOMACESSFILE_BUFFER_LENGTH];
			long m = length / b.length;// 读和写入2048个字节的次数
			int leftsize = (int) (length - m * b.length);// 剩下的字节数
			long timestart = System.currentTimeMillis();
			for (long i = 1; i <= m; i++) {
				fileAcessor.read(b);// 将数据从from文件读入
			}
			fileAcessor.read(b, 0, leftsize);// 读入最后剩下的字节
			fileAcessor.close();// 关闭输入
			long timeend = System.currentTimeMillis();
			if(logger.isDebugEnabled()){
				logger.info("线程" + Thread.currentThread().getName()
						+ "读写完毕,共使用" + (timeend - timestart) + "毫秒");
			}
			
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		return -1;
	}
	/**
	 * 
	 * @param file
	 * @param tempFile
	 * @param begin
	 * @param end
	 * @return
	 * @throws IOException
	 */
	  public  long writeTempFile(File file, File tempFile, long begin ,long end) throws IOException{
		  RandomAccessFile in=new RandomAccessFile(file,RANDOMACESSFILE_READ_MODE);
		  RandomAccessFile out=new RandomAccessFile(tempFile,RANDOMACESSFILE_WRITE_READ_MODE);
		  byte[] b=new byte[RANDOMACESSFILE_BUFFER_LENGTH];
		  int n=0;
		  in.seek(begin);
		  while(in.getFilePointer()<=end&&(n=in.read(b))!=-1){
			  out.write(b,0,n);
		  }
		  long endPointer=in.getFilePointer();
		  in.close();
		  out.close();
		  return endPointer;
	  }
	  
	
	public synchronized int read(byte[] buffer, int offset, int nLen) {
		int n = -1;
		FileInputStream is=null;
		try {
			 is=new FileInputStream(filePath);
			 //忽略前面的读取信息
			 is.skip(offset);
			 fileAcessor.seek(offset);
			 //定义读取输入流内容的缓存数据
			 byte[] buff=new byte[RANDOMACESSFILE_BUFFER_LENGTH];
			 //定义最多需要读取几次就可以完成本次的读取
			 long times=(nLen%RANDOMACESSFILE_BUFFER_LENGTH==0)?nLen/RANDOMACESSFILE_BUFFER_LENGTH:((nLen/RANDOMACESSFILE_BUFFER_LENGTH)+1);
			 int hasRead=0;
			 for (int i = 0; i < times; i++) {
				 hasRead=is.read(buff);
				 if(hasRead<0){
					 break;
				 }
				 fileAcessor.read(buffer);
			}
			n = nLen;
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			if(is!=null){
				try {
					is.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
				is=null;
			}
		}
		return n;
	}
	
	/**
	 * 关闭流对象
	 *
	 */
	public void close(){
	    if (fileAcessor != null) {
            try {
            	fileAcessor.close();
            } catch (Throwable ignore) {
            }
            fileAcessor = null;
        }
	}


	public String getFilePath() {
		return filePath;
	}

	public void setFilePath(String filePath) {
		this.filePath = filePath;
	}
	
	
}

 

 

package easyway.tbs.file.transport.core.split;

import java.io.File;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
 * 文件分割线程
 * @author longgangbai
 *
 */
public class FileSplitterFetch implements Runnable {
	 private final static Log logger = LogFactory.getLog(DataFileAccessor.class);
	  
    /**
     * 将读取到的字节输出到raf中
     */
    private DataFileAccessor raf;
    /**
     * 文件整个完成标识
     */
	private CountDownLatch masterLatch;
    /**
     * 当前文件读取执行完毕的标识
     */
	private CountDownLatch childrenLatch;
    /**
     * 文件分割的文件片段
     */
    private FileSplitterMessage message;
    
    

    /**
     * 构造器 
     * @param filepath 	服务器文件的路径
     * @param offset	输出流和读取起始点
     * @param length   输出流和读取结束点
     * @param buffer   数组的大小
     * @throws IOException
     */
    public FileSplitterFetch(FileSplitterMessage message,CountDownLatch masterLatch,CountDownLatch childrenLatch) throws IOException {
    	super();
    	this.message=message;
    	this.masterLatch=masterLatch;
    	this.childrenLatch=childrenLatch;
    	this.raf=new DataFileAccessor(message.getFilePath(),DataFileAccessor.RANDOMACESSFILE_READ_MODE);
    	System.out.println("raf=="+this.raf);
    }

    /**
     * 执行读取的方法
     */
    public void run() {

    	File tempFile = message.getTempFile();
    	try {
        	if(!tempFile.exists()){
        		tempFile.createNewFile();
        	}
			//masterLatch.await();
    		this.raf.writeTempFile(new File(message.getFilePath()), tempFile, message.getStartPos(), message.getEndPos());
        } catch (Exception ex) {
        	if(tempFile.exists()){
        		try {
					FileUtils.forceDelete(tempFile);
				} catch (IOException e) {
					
				}
        	}
        }finally {  
        	//使用finally块来关闭当前线程的输入流、输出流
        	this.raf.close();
        	//childrenLatch.countDown();
        }
        
    }


}

   

 

 

package easyway.tbs.file.transport.core.split;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 文件分割管理器
 * 当单个线程执行完毕之后，开始发送文件
 * @author longgangbai
 *
 */
public class FileSplitterManager {
	private static String File_Split_Separator="_";
	 private static final transient Logger logger = LoggerFactory.getLogger(FileSplitterManager.class);

	/**
	 * 文件的总长度
	 */
	private long fileLength;

	/**
	 * 整个文件分割完成的标识
	 * 开始执行发送消息
	 */
	private CountDownLatch masterLatch = new CountDownLatch(1);
	/**
	 *文件分割线程的数量
	 */
	private int splitThreadNum;
    /**
     * 
     * 当每一个线程执行完毕时候，则报告执行执行线程完成，当所有的执行线程
     * 执行完毕，则开始发送文件
     */
	private CountDownLatch childrenLatch = null;
	
	/**
	 * 文件片段的集合
	 */
	private List<FileSplitterMessage> fileSplitterMessageList=null;
	/**
	 * 创建文件锁
	 */
	private FileTransportLock fileLock=null;
	//发送文件的路径
	private String filepath;
	//发送文件的限制大小
	private int filepageSize;
	
	
	private String fileChannel;
	
	private File file;
	
	/**
	 * 
	 * @param filepath
	 * @param filepageSize
	 */
	public FileSplitterManager(String filepath,int filepageSize,String fileChannel){
		this.filepath=filepath;
		this.filepageSize=filepageSize;
		this.fileChannel=fileChannel;
	}

	/**
	 * 文件分割管理器的方法
	 *
	 */
	public void splitFileMgr(){
		file=new File(filepath);
		 //获取文件的大小
    	long fileTotalSize=file.length();
    	//如果文件的大小
    	if(filepageSize>0&&(filepageSize<fileTotalSize)){
    		initSplitFileMgr(file);
    	}else{
    		//直接发送文件
				fileSplitterMessageList=new ArrayList<FileSplitterMessage>(1);
				
				FileSplitterMessage fileSplitMessage=new  FileSplitterMessage();
	    		fileSplitMessage.setFilePath(file.getAbsolutePath());
	    		fileSplitMessage.setFileLength(file.length());
	    		fileSplitMessage.setFileName(file.getName());
	    		fileSplitMessage.setSplitThreadNum(1);
	    		fileSplitMessage.setCurrentThreadNum(1);
	    		fileSplitMessage.setStartPos(0);
	    		fileSplitMessage.setEndPos(filepageSize);
	    		fileSplitMessage.setLen(fileTotalSize);
	    		
	    		String threadNum="1";
	    		String currentThread="1";
	    		//临时文件的命名规则：源文件名称+“_”+"总线程数"+“_”+"当前线程编号"
	    		String tempFileName=file.getName()+File_Split_Separator+threadNum+File_Split_Separator+currentThread+".tmp";
	    		//文件相对比较小，文件的总线程数为1，当前线程也为1
	    		File tempFile=getTempFileName(file.getParent(),tempFileName);
	    		
	    		fileSplitMessage.setTempFile(tempFile);
	    		fileSplitterMessageList.add(fileSplitMessage);
	    		
	    		//重新命名原来的文件，标识文件已经开始发送
	    		file.renameTo(new File(file.getAbsolutePath()+".bak"));
	    		
	    		if(logger.isDebugEnabled()){
					logger.debug("the direct send file "+file.getAbsolutePath());
				}
    	}
    	System.out.println("fileSplitterMessageList ="+fileSplitterMessageList.size());
	}
	/**
	 * 创建临时文件
	 * @param parentPath
	 * @param tempFileName
	 * @return
	 */
	public File getTempFileName(String parentPath,String tempFileName){
		System.out.println(parentPath+File.separator+tempFileName);
			File tempFile=new File(parentPath+File.separator+tempFileName);
	
		try {
			if(tempFile.exists()){
				FileUtils.forceDelete(tempFile);
				if(logger.isDebugEnabled()){
					logger.info("文件存在，已经删除！");
				}
			}
			tempFile.createNewFile();
		} catch (IOException e) {
			if(logger.isDebugEnabled()){
				logger.info("文件创建失败："+e);
			}
		}
		return tempFile;
	}

	/**
	 * 分割文件的方法
	 *
	 */
	public void initSplitFileMgr(File file) {
		if(file.exists()){
			//创建文件锁
			//fileLock=new FileTransportLock(file,false);
			try {
				//设置文件锁
				//fileLock.lock();
				String fileName=file.getName();
	            //获取文件的长度
	        	fileLength=file.length();
	        	//获取文件对应的线程数
	        	splitThreadNum=(int)getThreadNum(fileLength,filepageSize);
	        	//线程同步线程
	        	childrenLatch = new CountDownLatch(splitThreadNum);
	        	//
	        	fileSplitterMessageList=new ArrayList<FileSplitterMessage>(splitThreadNum);
	        	//文件分割的过程
	        	splitFile(filepath,filepageSize,fileName);
			} catch (Exception e) {
				e.printStackTrace();
			}finally{
				//释放文件锁
				//fileLock.unlock();
			}
		}
	}
	

	/**
	 * 总的线程数
	 * @param fileSize
	 * @param filepageSize
	 * @return
	 */
	public long getThreadNum(long fileSize,long filepageSize){
		long 	threadNum=(fileSize%filepageSize==0)?(fileSize%filepageSize):((fileSize/filepageSize)+1);
		return threadNum;
	}
	
	/**
	 * 分解文件
	 * @param filepath  文件路径
	 * @param filepageSize  文件的大小
	 */ 
	public void splitFile(String filepath,int filepageSize,String fileName){
				
		        //当前线程真正读取的长度
	        	int length=filepageSize;
	        	//创建一个线程池
	        	ExecutorService threadExePool = Executors.newFixedThreadPool(splitThreadNum);
	        	//循环创建线程
	        	for(int i=1;i<=splitThreadNum;i++){
	        		//起始点
	        		int startPos=(i-1)*filepageSize;
	        		
	        		int endPos=i*filepageSize;
	        		
	        		FileSplitterMessage  fileSplitterMessage=new FileSplitterMessage();
	        		
	        		String tempFileName=file.getName()+File_Split_Separator+splitThreadNum+File_Split_Separator+i+".tmp";
	        		
	        		File tempFile=getTempFileName(file.getParent(), tempFileName);
	        		
	        		//针对文件最后一个片度结束点和片段长度
	        		if((splitThreadNum==i)&&(fileLength%filepageSize!=0)){
	        			endPos=(int)fileLength;
	        			length=(int)fileLength-startPos;
	        		}
	        		//设置文件片段内容
        			fileSplitterMessage.setTempFile(tempFile);
        			//文件读取起始点
        			fileSplitterMessage.setStartPos(startPos);
        			//文件读取起始点
        			fileSplitterMessage.setEndPos(endPos);
        			//文件总长度
        			fileSplitterMessage.setFileLength(fileLength);
        			//文件路径
        			fileSplitterMessage.setFilePath(filepath);
        			//文件分割线程数
        			fileSplitterMessage.setSplitThreadNum(splitThreadNum);
        			//当前线程编号数
        			fileSplitterMessage.setCurrentThreadNum(i);
        			//读取的长度
        			fileSplitterMessage.setLen(length);
        			
        			System.out.println("fileSplitterMessage="+fileSplitterMessage);
        			//文件的名称
        			fileSplitterMessage.setFileName(file.getName());
        			fileSplitterMessageList.add(fileSplitterMessage);
	        }
	       try {
	    		if(logger.isDebugEnabled()){
					logger.debug("文件"+fileName+"开始分割....");
				}
	    		//线程池开始执行线程
	        	for (FileSplitterMessage fileMessage : fileSplitterMessageList) {
	        		threadExePool.execute(new FileSplitterFetch(fileMessage,masterLatch,childrenLatch));					
				}
				masterLatch.countDown();//宣布开始
				if(logger.isDebugEnabled()){
					logger.debug("文件"+fileName+"分割完毕....");
				}
				//等待CountdownLatch信号为0，表示所有子线程都结束。 
				childrenLatch.await();//等待结束
		 
				System.out.println("size ================="+fileSplitterMessageList.size());
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				System.out.println("整个文件"+fileName+"分割完毕");
				//开始执行发送文件
			}
			//注意线程已经要结束了，但是threadExePool线程如果不关闭是不会结束的
			threadExePool.shutdown();
	}


	public int getFilepageSize() {
		return filepageSize;
	}


	public void setFilepageSize(int filepageSize) {
		this.filepageSize = filepageSize;
	}


	public String getFilepath() {
		return filepath;
	}


	public void setFilepath(String filepath) {
		this.filepath = filepath;
	}


	public String getFileChannel() {
		return fileChannel;
	}

	public void setFileChannel(String fileChannel) {
		this.fileChannel = fileChannel;
	}

}

 

 

 

package easyway.tbs.file.transport.core.split;

import java.io.File;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;

/**
 * 文件分割一段的信息
 * 
 * 
 * @author longgangbai
 *
 */
public class FileSplitterMessage {
	
		/**
		 * 文件路径
		 */
		private String filePath;
		/**
		 * 文件总长度
		 */
		private long fileLength;
		
		/**
		 * 文件名称
		 */
		private String fileName;
		/**
		 * 分割线程数量
		 */
		private int splitThreadNum;
		
		/**
		 * 当前分割线程编号
		 */
		private int currentThreadNum;
		/**
		 * 文件开始读取的起始点
		 */
		private int startPos;
		/**
		 * 文件开始读取的结束点
		 */
		private int endPos;
		/**
		 * 文件起止点之间的数据信息
		 */
		private byte[] data;
		/**
		 * 临时文件
		 */
		private File tempFile;
		/**
		 * 临时文件的长度
		 */
		private long len;
		
		
		
		public byte[] getData() {
			return data;
		}
		public void setData(byte[] data) {
			this.data = data;
		}
		public int getEndPos() {
			return endPos;
		}
		public void setEndPos(int endPos) {
			this.endPos = endPos;
		}
		public long getFileLength() {
			return fileLength;
		}
		public void setFileLength(long fileLength) {
			this.fileLength = fileLength;
		}
		public String getFilePath() {
			return filePath;
		}
		public void setFilePath(String filePath) {
			this.filePath = filePath;
		}
		public int getSplitThreadNum() {
			return splitThreadNum;
		}
		public void setSplitThreadNum(int splitThreadNum) {
			this.splitThreadNum = splitThreadNum;
		}
		public int getStartPos() {
			return startPos;
		}
		public void setStartPos(int startPos) {
			this.startPos = startPos;
		}
		public int getCurrentThreadNum() {
			return currentThreadNum;
		}
		public void setCurrentThreadNum(int currentThreadNum) {
			this.currentThreadNum = currentThreadNum;
		}
		

		public String getFileName() {
			return fileName;
		}
		public void setFileName(String fileName) {
			this.fileName = fileName;
		}
		public File getTempFile() {
			return tempFile;
		}
		public void setTempFile(File tempFile) {
			this.tempFile = tempFile;
		}
	
	    @Override
		public String toString(){
			return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
		}
		public long getLen() {
			return len;
		}
		public void setLen(long len) {
			this.len = len;
		}
}

 

 

 

package easyway.tbs.file.transport.core.split;

import java.io.File;

import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;

/**
 * 文件分割一段的信息
 * 
 * 
 * @author longgangbai
 *
 */
public class FileSplitterMessage {
	
		/**
		 * 文件路径
		 */
		private String filePath;
		/**
		 * 文件总长度
		 */
		private long fileLength;
		
		/**
		 * 文件名称
		 */
		private String fileName;
		/**
		 * 分割线程数量
		 */
		private int splitThreadNum;
		
		/**
		 * 当前分割线程编号
		 */
		private int currentThreadNum;
		/**
		 * 文件开始读取的起始点
		 */
		private int startPos;
		/**
		 * 文件开始读取的结束点
		 */
		private int endPos;
		/**
		 * 文件起止点之间的数据信息
		 */
		private byte[] data;
		/**
		 * 临时文件
		 */
		private File tempFile;
		/**
		 * 临时文件的长度
		 */
		private long len;
		
		
		
		public byte[] getData() {
			return data;
		}
		public void setData(byte[] data) {
			this.data = data;
		}
		public int getEndPos() {
			return endPos;
		}
		public void setEndPos(int endPos) {
			this.endPos = endPos;
		}
		public long getFileLength() {
			return fileLength;
		}
		public void setFileLength(long fileLength) {
			this.fileLength = fileLength;
		}
		public String getFilePath() {
			return filePath;
		}
		public void setFilePath(String filePath) {
			this.filePath = filePath;
		}
		public int getSplitThreadNum() {
			return splitThreadNum;
		}
		public void setSplitThreadNum(int splitThreadNum) {
			this.splitThreadNum = splitThreadNum;
		}
		public int getStartPos() {
			return startPos;
		}
		public void setStartPos(int startPos) {
			this.startPos = startPos;
		}
		public int getCurrentThreadNum() {
			return currentThreadNum;
		}
		public void setCurrentThreadNum(int currentThreadNum) {
			this.currentThreadNum = currentThreadNum;
		}
		

		public String getFileName() {
			return fileName;
		}
		public void setFileName(String fileName) {
			this.fileName = fileName;
		}
		public File getTempFile() {
			return tempFile;
		}
		public void setTempFile(File tempFile) {
			this.tempFile = tempFile;
		}
	
	    @Override
		public String toString(){
			return ToStringBuilder.reflectionToString(this, ToStringStyle.MULTI_LINE_STYLE);
		}
		public long getLen() {
			return len;
		}
		public void setLen(long len) {
			this.len = len;
		}
}

 

 

 

package easyway.tbs.file.transport.core.split;

/**
 * 分割文件测试类
 * @author longgangbai
 *
 */
public class FileSplitterManagerMain {
	
	public static void main(String[] args) {
		FileSplitterManager manager=new FileSplitterManager("E:/TestJoltProxy/RMAN.txt",10*1024*1024,"queue-svn");
		manager.splitFileMgr();
	}

}