Java实现多线程下载和断点续传

/** * 专注互联网,分享创造价值 * maoxiang@gmail.com * 2010-3-30下午04:40:06 */ package common.http; import java.io.BufferedInputStream; import java.io.ByteArrayOutputStream; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.io.RandomAccessFile; import java.util.concurrent.atomic.AtomicLong; import javax.net.ssl.SSLHandshakeException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.http.Header; import org.apache.http.HttpEntityEnclosingRequest; import org.apache.http.HttpHost; import org.apache.http.HttpRequest; import org.apache.http.HttpResponse; import org.apache.http.NoHttpResponseException; import org.apache.http.auth.AuthScope; import org.apache.http.auth.UsernamePasswordCredentials; import org.apache.http.client.HttpRequestRetryHandler; import org.apache.http.client.methods.HttpGet; import org.apache.http.conn.ClientConnectionManager; import org.apache.http.conn.params.ConnManagerParams; import org.apache.http.conn.params.ConnPerRoute; import org.apache.http.conn.params.ConnRoutePNames; import org.apache.http.conn.routing.HttpRoute; import org.apache.http.conn.scheme.PlainSocketFactory; import org.apache.http.conn.scheme.Scheme; import org.apache.http.conn.scheme.SchemeRegistry; import org.apache.http.impl.client.DefaultHttpClient; import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager; import org.apache.http.params.BasicHttpParams; import org.apache.http.params.CoreProtocolPNames; import org.apache.http.params.HttpParams; import org.apache.http.protocol.BasicHttpContext; import org.apache.http.protocol.ExecutionContext; import org.apache.http.protocol.HttpContext; /** * 一个多线程支持断点续传的工具类<br/> * 2010-03 用Htpp Component重写 */ public class HttpDownloader1 { private final Log log = LogFactory.getLog(getClass().getName()); private int threads = 5; // 总共的线程数 private int maxThreads = 10; // 最大的线程数 private String destUrl; // 目标的URL private String savePath; // 保存的路径 private File lockFile;// 用来保存进度的文件 private String userAgent = "jHttpDownload"; private boolean useProxy = false; private String proxyServer; private int proxyPort; private String proxyUser; private String proxyPassword; private int blockSize = 1024 * 4; // 4K 一个块 // 1个位代表一个块,，用来标记是否下载完成 private byte[] blockSet; private int blockPage; // 每个线程负责的大小 private int blocks; private boolean running; // 是否运行中,避免线程不能释放 private DefaultHttpClient httpClient;// // =======下载进度信息 private long beginTime; private AtomicLong downloaded = new AtomicLong(0); // 已下载的字节数\ private long fileLength; // 总的字节数 // 监控线程,用来保存进度和汇报进度 private MonitorThread monitorThread = new MonitorThread(); public HttpDownloader1(String destUrl, String savePath, int threads) { this.threads = threads; this.destUrl = destUrl; this.savePath = savePath; } public HttpDownloader1(String destUrl, String savePath) { this(destUrl, savePath, 5); } /** * 开始下载 */ public boolean download() { log.info("下载文件" + destUrl + ",保存路径=" + savePath); beginTime = System.currentTimeMillis(); boolean ok = false; try { File saveFile = new File(savePath); lockFile = new File(savePath + ".lck"); if (lockFile.exists() && !lockFile.canWrite()) { throw new Exception("文件被锁住，或许已经在下载中了"); } File parent = saveFile.getParentFile(); if (!parent.exists()) { log.info("创建目录=" + parent.getAbsolutePath()); } if (!parent.canWrite()) { throw new Exception("保存目录不可写"); } if (saveFile.exists()) { if (!saveFile.canWrite()) { throw new Exception("保存文件不可写,无法继续下载"); } log.info("检查之前下载的文件"); if (lockFile.exists()) { log.info("加载之前下载进度"); loadPrevious(); } } else { lockFile.createNewFile(); } // 1初始化httpClient setupHttpClient(); HttpResponse response = getResponse(0); Header length = response.getFirstHeader("Content-Length"); if (length != null) { try { fileLength = Long.parseLong(length.getValue()); } catch (Exception e) { } } log.info("下载文件的大小:" + fileLength); if (fileLength <= 0) { // 不支持多线程下载,采用单线程下载 log.info("服务器不能返回文件大小，采用单线程下载"); threads = 1; } if (response.getFirstHeader("Content-Range") == null) { log.info("服务器不支持断线续传"); threads = 1; } else { log.info("服务器支持断点续传"); } if (blockSet != null) { log.info("检查文件，是否能够续传"); if (blockSet.length * 8l * blockSize < fileLength) { log.info("文件大小已改变，需要重新下载"); blockSet = null; } } // if (fileLength > 0 && parent.getFreeSpace() < fileLength) { throw new Exception("磁盘空间不够"); } if (fileLength > 0) { int i = (int) (fileLength / blockSize); if (fileLength % blockSize > 0) { i++; } blocks = i; log.info("文件的块数:" + blocks); blockSet = BitUtil.createBit(blocks); } else { // 一个块 blocks = 1; } blockPage = blocks / threads; // 每个线程负责的块数 log.info("分配线程。线程数量=" + threads + ",块总数=" + blocks + ",总字节数=" + fileLength + ",每块大小=" + blockSize + ",块/线程=" + blockPage); // 检查 running = true; ThreadGroup downloadGroup = new ThreadGroup("download"); for (int i = 0; i < threads; i++) { int begin = i * blockPage; int end = (i + 1) * blockPage; if (i == threads - 1 && blocks % threads > 0) { // 如果最后一个线程，有余数，需要修正 end = blocks; } // 扫描每个线程的块是否有需要下载的 boolean needDownload = false; for (int j = begin; j < end; j++) { if (!BitUtil.getBit(blockSet, j)) { needDownload = true; break; } } if (!needDownload) { log.info("所有块已经下载完毕.Begin=" + begin + ",End=" + end); } // 启动下载其他线程 DownloadThread downloadThread = new DownloadThread( downloadGroup, i, begin, end); downloadThread.start(); } monitorThread.setStop(false); monitorThread.start(); while (downloadGroup.activeCount() > 0) { Thread.sleep(2000); } ok = true; } catch (Exception e) { e.printStackTrace(); log.error(e); } finally { // closeHttpClient(); if (ok) { log.info("删除进度文件:" + lockFile.getAbsolutePath()); lockFile.delete(); } httpClient = null; } monitorThread.setStop(true); log.info("下载完成，耗时:" + getTime((System.currentTimeMillis() - beginTime) / 1000)); return ok; } private void loadPrevious() throws Exception { ByteArrayOutputStream outStream = new ByteArrayOutputStream(); FileInputStream inStream = new FileInputStream(lockFile); byte[] buffer = new byte[4096]; int n = 0; while (-1 != (n = inStream.read(buffer))) { outStream.write(buffer, 0, n); } outStream.close(); inStream.close(); blockSet = outStream.toByteArray(); log.debug("之前的文件大小应该是:" + blockSet.length * 8l * blockSize); } private void setupHttpClient() throws Exception { HttpParams params = new BasicHttpParams(); SchemeRegistry schemeRegistry = new SchemeRegistry(); schemeRegistry.register(new Scheme("http", PlainSocketFactory .getSocketFactory(), 80)); ClientConnectionManager cm = new ThreadSafeClientConnManager(params, schemeRegistry); httpClient = new DefaultHttpClient(cm, params); httpClient.getParams().setParameter(CoreProtocolPNames.USER_AGENT, userAgent); ConnPerRoute defaultConnPerRoute = new ConnPerRoute() { public int getMaxForRoute(HttpRoute