以多线程、断点续传方式下载文件的实现

以多线程、断点续传方式下载文件，经常出现下载下来的文件大小和服务端一致，但是却无法正常打开的现象，搞了很久，贴下我的实现方式，请各位多多指教

思路：

1、将下载文件的处理放在自定义的线程类中，每下载一个文件就新启动一个下载线程。

2、在下载线程中完成对服务端的链接和身份认证，成功后开始下载文件。

3、新建n个子线程，根据下载文件的大小和线程数量得到每个子线程要下载的大小。

4、分别启动子线程，进行分段下载。

5、分段下载完成，合并临时文件。

6、合并文件完成，删除临时文件。

 

实现：

FTP下载线程类

package com.jfc.ftp.tools;  
  
import java.io.BufferedInputStream;  
import java.io.BufferedOutputStream;  
import java.io.File;  
import java.io.FileInputStream;  
import java.io.FileOutputStream;  
import java.util.concurrent.locks.Lock;  
import java.util.concurrent.locks.ReentrantLock;  
  
import com.jfc.ftp.service.FTPService;  
import com.jfc.ftp.util.Constant;  
import com.jfc.ftp.util.PropertyUtil;  
  
/** 
 * 为断点续传下载文件而启动新的线程 
 * @author SavageGarden 
 * 
 */  
public class FTPThread extends Thread{  
    /** 
     * 站点URL 
     */  
    private String host;  
    /** 
     * 站点端口 
     */  
    private int port;  
    /** 
     * 用户 
     */  
    private String user;  
    /** 
     * 密码 
     */  
    private String pswd;  
    /** 
     * 当前线程的FTP操作接口实现类 
     */  
    private FTPService ftpService;  
  
    /** 
     * 第几个下载项 
     */  
    private int rowIndex;  
    /** 
     * 要下载的文件路径 
     */  
    private String filepath;  
    /** 
     * 要下载的文件大小 
     */  
    private long filesize;  
    /** 
     * 要下载的文件保存路径 
     */  
    private String savepath;  
      
    /** 
     * 标记文件已下载量 
     */  
    public int hadRead = 0;  
    /** 
     * 下载线程开始时间 
     */  
    public long startTime = 0;  
    /** 
     * 下载线程结束时间 
     */  
    public long endTime = 0;  
    /** 
     * 当前下载线程的互斥锁 
     */  
    public Lock ftpThreadLock;  
    /** 
     * 当前下载线程的状态 
     */  
    private int status = Constant.THREAD_STATUS_NEW;  
      
    public synchronized int getStatus() {  
        return status;  
    }  
    public synchronized void setStatus(int status) {  
        this.status = status;  
    }  
    /** 
     * 是否已经合并文件 
     */  
    private boolean hadMerger = false;  
      
    public synchronized boolean isHadMerger() {  
        return hadMerger;  
    }  
    public synchronized void setHadMerger(boolean hadMerger) {  
        this.hadMerger = hadMerger;  
    }  
    /** 
     * 当前下载线程的状态 
     */  
    private int completed = 0;  
      
    public synchronized int getCompleted() {  
        return completed;  
    }  
    public synchronized void setCompleted(int completed) {  
        this.completed = completed;  
    }  
    /** 
     * 下载线程个构造方法<br> 
     * 根据已经取得连接的FTPTools得到连接信息<br> 
     * 根据参数取得下载信息 
     * @param rowIndex 
     * @param filepath 
     * @param filesize 
     * @param savepath 
     */  
    public FTPThread(int rowIndex, String filepath, long filesize, String savepath) {  
        super("FTPThread");  
        host = FTPTools.host;  
        port = FTPTools.port;  
        user = FTPTools.user;  
        pswd = FTPTools.pswd;  
        this.rowIndex = rowIndex;  
        this.filepath = filepath;  
        this.filesize = filesize;  
        this.savepath = savepath;  
        ftpThreadLock = new ReentrantLock();  
        setStatus(Constant.THREAD_STATUS_RUNNABLE);  
        start();  
    }  
    public FTPThread(int rowIndex, String filepath, long filesize, String savepath, int status) {  
        super("FTPThread");  
        host = FTPTools.host;  
        port = FTPTools.port;  
        user = FTPTools.user;  
        pswd = FTPTools.pswd;  
        this.rowIndex = rowIndex;  
        this.filepath = filepath;  
        this.filesize = filesize;  
        this.savepath = savepath;  
        ftpThreadLock = new ReentrantLock();  
        setStatus(status);  
        start();  
    }  
    public void run() {  
        getFTPService();  
        getFTPConnect(host, port);  
        if(doLoginFTP(user, pswd)) {  
            ResumeBrokenTransferByThread(this, rowIndex, filepath, filesize, savepath);  
        }  
    }  
    /** 
     * 获取FTPService接口实现类<br> 
     * 首先从配置文件中找<br> 
     * 如果没有则加载默认的实现类 
     * @return 
     * @throws InstantiationException 
     * @throws IllegalAccessException 
     * @throws ClassNotFoundException 
     */  
    public void getFTPService(){  
        try {  
            ftpService = (FTPService)Class.forName(PropertyUtil.getProperty("ftp.service.name", FTPService.FTP_SERVICE_NAME)).newInstance();  
        } catch (Exception e) {  
            e.printStackTrace();  
        }   
    }  
    /** 
     * 根据服务器地址、端口获得ftp链接 
     * @param host 
     * @param port 
     * @return 
     */  
    public String getFTPConnect(String host, int port) {  
        return ftpService.getFTPConnect(host, port);  
    }  
    /** 
     * 执行登录 
     * @param user 
     * @param pswd 
     * @return 
     */  
    public boolean doLoginFTP(String user, String pswd) {  
        return ftpService.doLoginFTP(user, pswd);  
    }  
    /** 
     * 以断点续传的方式下载文件 
     * @param rowIndex 
     * @param filepath 
     * @param filesize 
     * @param savepath 
     */  
    public void ResumeBrokenTransfer(int rowIndex, String filepath, int filesize, String savepath) {  
        ftpService.ResumeBrokenTransfer(rowIndex, filepath, filesize, savepath);  
    }  
    /** 
     * 以多线程、断点续传的方式下载文件 
     * @param rowIndex 
     * @param filepath 
     * @param filesize 
     * @param savepath 
     */  
    public void ResumeBrokenTransferByThread(FTPThread ftpThread, int rowIndex, String filepath, long filesize, String savepath) {  
        ftpService.ResumeBrokenTransferByThread(ftpThread, rowIndex, filepath, filesize, savepath);  
    }  
    /** 
     * 在指定文件路径下查找临时文件合并为一个文件 
     * @param filepath 
     * @param threadCount 
     */  
    public void mergerTempFile(String filepath, int threadCount) {  
        System.out.println("开始合并文件");  
        try {  
            BufferedOutputStream data_output = new BufferedOutputStream(new FileOutputStream(filepath));  
            byte[] temp = new byte[Constant.TEMP_BYTE_LENGTH];  
            for(int i = 0; i < threadCount; i ++) {  
                File tempFile = new File(filepath + Constant.DOWNLOAD_TEMP_NAME + i);  
                BufferedInputStream data_input = new BufferedInputStream(new FileInputStream(tempFile));  
                int read = 0;  
                int hadRead = 0;  
                while((read = data_input.read(temp, 0, temp.length)) != -1) {  
                    data_output.write(temp, 0, read);  
                    hadRead += read;  
                }  
                data_input.close();  
            }  
            data_output.close();  
        } catch (Exception e) {  
            e.printStackTrace();  
        }   
        setHadMerger(true);  
        System.out.println("合并文件完成");  
        deleteTempFile(filepath, threadCount);  
    }  
    /** 
     * 合并文件完成后删除临时文件 
     * @param filepath 
     * @param threadCount 
     */  
    public void deleteTempFile(String filepath, int threadCount) {  
        if(isHadMerger()) {  
            for(int i = 0; i < threadCount; i ++) {  
                File tempFile = new File(filepath + Constant.DOWNLOAD_TEMP_NAME + i);  
                tempFile.delete();  
            }  
        }  
    }  
}  

 

FTP接口实现类中的ResumeBrokenTransferByThread方法

 

/** 
     * 使用多线程、断点续传方式下载文件<br> 
     * 首先查找要保存的路径下有无缓存文件(以.wfml为后缀)<br> 
     *      不存在 重新开始下载<br> 
     *      存在      继续<br> 
     * 重新开始下载<br> 
     *      读取配置文件要分多少个线程来下载文件<br> 
     *      按照文件大小、线程数量来分配每个线程要下载的文件大小、文件名<br> 
     *      开始断点续传下载<br> 
     * 继续下载<br> 
     *      开始断点续传下载<br> 
     * @param rowIndex 
     * @param filepath 
     * @param filesize 
     * @param savepath 
     */  
    public void ResumeBrokenTransferByThread(final FTPThread ftpThread, final int rowIndex, final String filepath, final long filesize, final String savepath) {  
        final String filename = filepath.substring(filepath.lastIndexOf("/") + 1, filepath.length());  
        <span style="color: #ff0000;">final byte[] temp = new byte[Constant.TEMP_BYTE_LENGTH];</span>  
  
        //得到要创建的线程数量  
        final int threadCount = Integer.parseInt(PropertyUtil.getProperty(Constant.RESUME_THREAD_COUNT_PROPNAME, Constant.RESUME_THREAD_COUNT_DEFAULT));  
        final String[] downloadSizeArray = SystemTools.getDownloadSizeArray(filesize, threadCount);  
        for(int i = 0; i < threadCount; i ++) {  
            File temp_file = new File(savepath + File.separator + filename + Constant.DOWNLOAD_TEMP_NAME + i);  
            System.out.println("文件" + i + "大小为：" + temp_file.length());  
            ftpThread.hadRead += temp_file.length();  
              
        }  
        System.out.println("ftpThread.hadRead : " + ftpThread.hadRead);  
        for(int i = 0; i < threadCount; i ++) {  
            final int index = i;  
            Thread resumeThread = new Thread(){  
                //当前线程的缓存文件  
                File tempFile = new File(savepath + File.separator + filename + Constant.DOWNLOAD_TEMP_NAME + index);  
                public void run() {  
                    SocketFTPService socketFTPService = new SocketFTPService();  
                    socketFTPService.getFTPConnect(host, port);  
                    if(socketFTPService.doLoginFTP(user, pswd)) {  
                        try {  
                            int read = 0;  
                            int hadRead = 0;  
                            //当前线程要下载的文件大小  
                            String downsize = downloadSizeArray[index].split(":")[1];  
                            //当前线程要下载的文件起始点  
                            String skipsize = downloadSizeArray[index].split(":")[0];  
                            //将hadRead(已读文件大小)置为缓存文件的大小  
                            hadRead = (int)tempFile.length();  
                            //设定文件指针(下载位置)  
                            //socketFTPService.doFTPCommand("REST " + (Integer.parseInt(skipsize) + temp_file.length()));  
                            //readRespond();  
                            socketFTPService.dataSocket = socketFTPService.getDataSocket();  
                            socketFTPService.doFTPCommand("RETR " + filepath);  
                            BufferedInputStream data_input = new BufferedInputStream(socketFTPService.dataSocket.getInputStream());  
                              
                            RandomAccessFile raf = new RandomAccessFile(tempFile, "rw");  
                            //跳过当前线程要下载的文件起始点和缓存文件大小之和  
                            data_input.skip(Integer.parseInt(skipsize) + hadRead);  
                            raf.seek(hadRead);  
                              
                            System.out.println("线程" + index + "已下载 " + hadRead);  
                            if(ftpThread.startTime == 0) {  
                                ftpThread.startTime = System.currentTimeMillis();  
                            }  
                            SystemTools.addObserver();  
                              
                            while(hadRead < Integer.parseInt(downsize)) {  
                                if(ftpThread.getStatus() == Constant.THREAD_STATUS_RUNNABLE) {  
                                    while((read = data_input.read(temp, 0, temp.length)) != -1) {  
                                        int temp_hadRead = hadRead;  
                                        if((temp_hadRead += read) > Integer.parseInt(downsize)) {  
                                            read = Integer.parseInt(downsize) - hadRead;  
                                        }  
                                        raf.write(temp, 0, read);  
                                        hadRead += read;  
                                          
                                        ftpThread.ftpThreadLock.lock();  
                                        try {  
                                            ftpThread.hadRead += read;  
                                        } finally {  
                                            ftpThread.ftpThreadLock.unlock();  
                                        }  
                                        SystemTools.getCurrentSpeed(rowIndex, ftpThread.startTime, ftpThread.hadRead);  
                                        SystemTools.getPrice(rowIndex, ftpThread.hadRead, filesize);  
                                        SwingUtilities.invokeLater(SystemTools.updateProgressBarRunnable);  
                                    }  
                                    System.out.println("第" + index + "个线程完成下载" + hadRead + ",完成下载" + ftpThread.hadRead);  
                                    raf.close();  
                                    if(hadRead == tempFile.length()) {  
                                        ftpThread.setCompleted(ftpThread.getCompleted() + 1);  
                                        System.out.println(ftpThread.getCompleted());  
                                    }  
                                    if(ftpThread.getCompleted() == threadCount && ftpThread.hadRead == filesize) {  
                                        ftpThread.endTime = System.currentTimeMillis();  
                                        SystemTools.getFinalSpeed(rowIndex, ftpThread.startTime, ftpThread.endTime, filesize);  
                                        ftpThread.mergerTempFile(savepath + File.separator + filename, threadCount);  
                                    }  
                                }  
                            }  
                        } catch(Exception e) {  
                            e.printStackTrace();  
                        }  
                    }  
                }  
            };  
            resumeThread.start();  
        }  
    }  

 

 

摘自：http://www.iteye.com/topic/553932