Java Nio 多线程网络下载

--> 默认最多50个线程 同一文件下载失败延迟超过30秒就结束下载

--> 下载5分钟超时时间，假设5分钟内未下载完就结束下载

--> 依赖 commons-httpclient 与 commons-io 包

package com.leunpha;import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.params.HttpClientParams;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;import java.io.*;
import java.lang.reflect.Method;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Observable;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.zip.ZipFile;/**
* User: zhoujingjie
* Date: 14-4-18
* Time: 下午12:52
*/
public class Downloader extends Observable {
protected String url, savePath; //下载地址与保存路径
protected FileChannel channel; //保存文件的通道
protected long size, perSize; //文件大小与每一个小文件的大小
protected volatile long downloaded; // 已下载的
protected int connectCount; //连接数
protected Connection[] connections; //连接对象
protected boolean isSupportRange; //是否支持断点下载
protected long timeout; //超时
protected boolean exists; //是否存在
private RandomAccessFile randomAccessFile;
protected volatile boolean stop; //停止
private static volatile boolean exception; //是否异常
private AtomicLong prevDownloaded = new AtomicLong(0); //上一次的下载结果
private static Log log = LogFactory.getLog(Downloader.class);
private AtomicInteger loseNum = new AtomicInteger(0);
private int maxThread; public Downloader(String url, String savePath) throws IOException {
//超时一小时
this(url, savePath, 1000 * 60*5,50);
} public Downloader(String url, String savePath, long timeout,int maxThread) throws FileNotFoundException {
this.timeout = timeout;
this.url = url;
this.maxThread = maxThread;
File file = new File(savePath);
if (!file.exists()) file.mkdirs();
this.savePath= file.getAbsolutePath() + "/" + url.substring(url.lastIndexOf("/"));
exists = new File(this.savePath).exists();
if(!exists){
randomAccessFile= new RandomAccessFile(this.savePath+".temp", "rw");
channel =randomAccessFile.getChannel();
}
} public GetMethod method(long start, long end) throws IOException {
GetMethod method = new GetMethod(Downloader.this.url);
method.setRequestHeader("User-Agent", "Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/30.0.1599.101 Safari/537.36");
if (end > 0) {
method.setRequestHeader("Range", "bytes=" + start + "-" + (end - 1));
} else {
method.setRequestHeader("Range", "bytes=" + start + "-");
}
HttpClientParams clientParams = new HttpClientParams();
//5秒超时
clientParams.setConnectionManagerTimeout(5000);
HttpClient client = new HttpClient(clientParams);
client.executeMethod(method);
int statusCode = method.getStatusCode();
if (statusCode >= 200 && statusCode < 300) {
isSupportRange = (statusCode == 206) ? true : false;
}
return method;
} public void init() throws IOException {
size = method(0, -1).getResponseContentLength();
if (isSupportRange) {
if (size < 4 * 1024 * 1024) { //假设小于4M
connectCount = 1;
} else if (size < 10 * 1024 * 1024) { //假设文件小于10M 则两个连接
connectCount = 2;
} else if (size < 30 * 1024 * 1024) { //假设文件小于80M 则使用6个连接
connectCount = 3;
} else if (size < 60 * 1024 * 1024) { //假设小于60M 则使用10个连接
connectCount = 4;
} else {
//否则为10个连接
connectCount = 5;
}
} else {
connectCount = 1;
}
log.debug(String.format("%s size:%s connectCount:%s", this.url, this.size, this.connectCount));
perSize = size / connectCount;
connections = new Connection[connectCount];
long offset = 0;
for (int i = 0; i < connectCount - 1; i++) {
connections[i] = new Connection(offset, offset + perSize);
offset += perSize;
}
connections[connectCount - 1] = new Connection(offset, size);
} /**
* 强制释放内存映射
*
* @param mappedByteBuffer
*/
static void unmapFileChannel(final MappedByteBuffer mappedByteBuffer) {
try {
if (mappedByteBuffer == null) {
return;
}
mappedByteBuffer.force();
AccessController.doPrivileged(new PrivilegedAction<Object>() {
@Override
public Object run() {
try {
Method getCleanerMethod = mappedByteBuffer.getClass().getMethod("cleaner", new Class[0]);
getCleanerMethod.setAccessible(true);
sun.misc.Cleaner cleaner = (sun.misc.Cleaner) getCleanerMethod.invoke(mappedByteBuffer, new Object[0]);
cleaner.clean();
} catch (Exception e) {
//LOG.error("unmapFileChannel." + e.getMessage());
}
return null;
}
});
} catch (Exception e) {
log.debug("异常->exception=true");
exception = true;
log.error(e);
}
} private void timer() {
Timer timer = new Timer();
//延迟3秒，3秒执行一次
timer.schedule(new TimerTask() {
@Override
public void run() {
log.debug(String.format("已下载-->%s -> %s",(((double) downloaded) / size * 100) + "%", url));
//假设上一次的下载大小与当前的一样就退出
if(prevDownloaded.get() ==downloaded && downloaded<size){
if(loseNum.getAndIncrement()>=10){
log.debug(String.format("上次下载%s与当前下载%s一致,exception->true url:%s ",prevDownloaded.get(),downloaded,url));
exception = true;
}
}
//假设下载完毕或者异常就退出
if(downloaded>=size || exception){
stop = true;
cancel();
}
//设置上次下载的大小等于如今的大小
prevDownloaded.set(downloaded);
}
},3000,3000);
} public void start() throws IOException {
if (exists) {
log.info("文件已存在." + this.url);
Thread.currentThread().interrupt();
return;
}
while (Thread.activeCount()>maxThread){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {}
}
init();
timer();
CountDownLatch countDownLatch = new CountDownLatch(connections.length);
log.debug("開始下载:" + url);
for (int i = 0; i < connections.length; i++) {
new DownloadPart(countDownLatch, i).start();
}
end(countDownLatch);
} private boolean rename(File tempFile){
File file = new File(this.savePath);
boolean isRename=tempFile.renameTo(file);
if(!isRename){
try {
IOUtils.copy(new FileInputStream(tempFile),new FileOutputStream(file));
} catch (IOException e) {
log.error(e);
}
}
return true;
} public void end(CountDownLatch countDownLatch){
try {
//超过指定时间就直接结束
countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
exception = true;
log.error(e);
log.info("下载失败:"+this.url);
} finally {
try {
channel.force(true);
channel.close();
randomAccessFile.close();
} catch (IOException e) {
log.error(e);
}
File temp = new File(this.savePath+".temp");
log.debug(String.format("%s %s", exception, this.url));
//假设有异常则删除已下载的暂时文件
if(exception){
if(!temp.delete()){
if(temp!=null)temp.delete();
}
}else{
try {
Thread.sleep(100);
} catch (InterruptedException e) {}
rename(temp);
setChanged();
notifyObservers(this.url);
log.info("下载成功:"+this.url);
}
}
} private class Connection {
long start, end; public Connection(long start, long end) {
this.start = start;
this.end = end;
} public InputStream getInputStream() throws IOException {
return method(start, end).getResponseBodyAsStream();
}
} private class DownloadPart implements Runnable {
CountDownLatch countDownLatch;
int i; public DownloadPart(CountDownLatch countDownLatch, int i) {
this.countDownLatch = countDownLatch;
this.i = i;
}
public void start() {
new Thread(this).start();
}
@Override
public void run() {
MappedByteBuffer buffer = null;
InputStream is = null;
try {
is = connections[i].getInputStream();
buffer = channel.map(FileChannel.MapMode.READ_WRITE, connections[i].start, connections[i].end - connections[i].start);
byte[] bytes = new byte[4 * 1024];
int len;
while ((len = is.read(bytes)) != -1 && !exception && !stop) {
buffer.put(bytes, 0, len);
downloaded+= len;
}
log.debug(String.format("file block had downloaded.%s %s",i,url));
} catch (IOException e) {
log.error(e);
} finally {
unmapFileChannel(buffer);
if(buffer != null)buffer.clear();
if (is != null) try {
is.close();
} catch (IOException e) {
}
countDownLatch.countDown();
}
}
}}