用Java实现多线程下载

计划做一个多线程下载的程序，强化一下对多线程的理解。计划第一步先做出一个多线程下载的基本功能，下一步做出断点续传等功能。

1.确定下载核心类的接口，接口代码如下，之前看过一大牛的代码，断点续传写得很不错，但是把URL中文件名的解析等工作都放到了下载类中，我觉得这样会导致这个类多担负了一个任务，而我这里，将URL中文件名的解析放到Client中去做，而Client可能是字符型的客户端，或者GUI，这都不会影响到下载核心类的实现。Client仅依赖于DownloadService这一接口，而DownloadService的具体实现可以变化而不影响Client。比如，我实现了两种DownloadService，一种 单线程，一种多线程的。客户端的代码无需改动。

/**
 * 下载服务接口
 * 提供下载的基本功能，包括下载一个文件，获取当前进度以及断点续传等
 * @author without me
 *
 */
public interface DownloadService
{
	/**
	 * 下载地址为url的一个文件
	 * @param url 文件的URL地址
	 * @param saveFile 保存的文件
	 */
	public void download(URL url, File saveFile) throws Exception;
	
	/**
	 * 获取当前下载文件大小
	 * @return 返回当前下载文件的字节数
	 */
	public long getFileSize();
	
	/**
	 * 获取已经下载的文件的大小
	 * @return 返回已经下载部分的字节数
	 */
	public long getFinishedSize();
	
	/**
	 * 获取当前下载完成的比率
	 * @return 当前完成的百分比
	 */
	public double getFinishedRate();
	
	/**
	 * 断点续传，继续下载地址为url的文件
	 * @param url 续传地址为url的文件
	 * @param tmpFile 完成部分下载的临时文件
	 */
	public void continueDownload(String url, File tmpFile);
	
	/**
	 * 暂停下载
	 */
	public void pause() throws InterruptedException;
	
	/**
	 * 继续下载
	 */
	public void resume() throws InterruptedException;
	
	/**
	 * 当前下载服务是否处于暂停状态
	 * @return 处于暂停状态返回true，否则false
	 */
	public boolean isPasued();
}

2. 先实现单线程的下载，代码非常简单，如下。其中断点续传功能未实现。

public class SingleThreadLoader implements DownloadService
{
	private long finishedBytes;
	private long totalBytes;
	private static final int BUFFER_SIZE = 1024;
	private boolean isPaused = false;
	private boolean pause = false;
	@Override
	public void continueDownload(String url, File tmpFile)
	{
	}
	@Override
	public void download(URL url, File saveFile) throws IOException, InterruptedException
	{
		HttpURLConnection con = null;
		InputStream inStream = null;
		OutputStream outStream = null;
		try
		{
			con = (HttpURLConnection) url.openConnection();
			setHeader(con);
			con.setConnectTimeout(10000);
			con.setReadTimeout(10000);
			totalBytes = con.getContentLength();
			inStream = con.getInputStream();
			System.out.println(con.getResponseCode());
			outStream = new FileOutputStream(saveFile);
			byte[] buffer = new byte[BUFFER_SIZE];
			finishedBytes = 0;
			int len = 0;
			while ((len = inStream.read(buffer)) > 0)
			{
				finishedBytes += len;
				outStream.write(buffer, 0, len);
				if (pause)
				{
					synchronized (this)
					{
						this.isPaused = true;
						this.wait();
					}
				}
			}
		}
		finally
		{
			if (inStream != null)
				inStream.close();
			if (outStream != null)
				outStream.close();
			if (con != null)
				con.disconnect();
		}
	}
	private static boolean setHeader(URLConnection con)
	{
		Properties prop = new Properties();
		try
		{
			prop.load(new FileInputStream("header.properties"));
		}
		catch (FileNotFoundException e)
		{
			e.printStackTrace();
			return false;
		}
		catch (IOException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		for (Iterator iter = prop.keySet().iterator(); iter.hasNext();)
		{
			String key = (String) iter.next();
			con.setRequestProperty(key, prop.getProperty(key));
		}
		return true;
	}
	@Override
	public long getFileSize()
	{
		return totalBytes;
	}
	@Override
	public double getFinishedRate()
	{
		return finishedBytes / (double) totalBytes;
	}
	@Override
	public long getFinishedSize()
	{
		return finishedBytes;
	}
	@Override
	public void pause() throws InterruptedException
	{
		this.pause = true;
	}
	@Override
	public void resume() throws InterruptedException
	{
		this.isPaused = false;
		this.pause = false;
		synchronized (this)
		{
			this.notifyAll();
		}
	}
	@Override
	public boolean isPasued()
	{
		return isPaused;
	}
}

其中需要用到连接的header设置文件header.properties，如下：

User-Agent=Mozilla/5.0 (X11; U; Linux i686; en-US; rv:1.9.0.3) Gecko/2008092510 Ubuntu/8.04 (hardy) Firefox/3.0.3
Accept-Language=en-us,en;q=0.7,zh-cn;q=0.3
Accept-Encoding=aa
Accept-Charset=ISO-8859-1,utf-8;q=0.7,*;q=0.7
Keep-Alive=600
Connection=keep-alive
If-Modified-Since=Fri, 02 Jan 1980 17:00:05 GMT
If-None-Match=/"1261d8-4290-df64d224/"
Cache-Control=max-age=0
Referer=www.google.com

3. 实现多线程下载，代码如下：

public class MultiThreadLoader implements DownloadService
{
	private URL url;
	private File saveFile;
	private volatile long finishedBytes;
	private volatile long totalBytes;
	private CountDownLatch latch;
	private int numOfThreads;
	private static final int BUFFER_SIZE = 1024;
	public MultiThreadLoader()
	{
	}
	public MultiThreadLoader(URL url, File saveFile, int numOfThreads)
	{
		this.numOfThreads = numOfThreads;
	}
	@Override
	public void continueDownload(String url, File tmpFile)
	{
	}
	private static class ChildDownloadThread implements Runnable
	{
		private long startPos;
		private long endPos;
		private int tryTimes = 0;
		private MultiThreadLoader loader;
		public ChildDownloadThread(MultiThreadLoader loader, long startPos, long endPos)
		{
			this.startPos = startPos;
			this.endPos = endPos;
			this.loader = loader;
		}
		public void run()
		{
			InputStream inStream = null;
			RandomAccessFile randAccFile = null;
			try
			{
				HttpURLConnection con = (HttpURLConnection) loader.url.openConnection();
				setHeader(con);
				con.setConnectTimeout(10000);
				con.setReadTimeout(10000);
				setPosHeader(con, startPos, endPos);
				randAccFile = new RandomAccessFile(loader.saveFile, "rw");
				randAccFile.seek(startPos);
				inStream = con.getInputStream();
				byte[] buffer = new byte[BUFFER_SIZE];
				int len = 0;
				long localFinished = startPos;
				int localRead = 0;
//				System.out.println("Thread" + Thread.currentThread().getId() + " startPos=" + startPos + ", endPos=" + endPos + " ResponseCode=" + con.getResponseCode());
				while (true)
				{
					try
					{
						len = inStream.read(buffer);
						localRead += len;
					}
					catch (SocketTimeoutException e)
					{
						tryTimes ++;
//						System.out.println(Thread.currentThread().getId() + "SocketTimeoutException occured, tryed " + tryTimes + " times.");
						continue;
					}
					if(len < 0)
					{
//						System.out.println(Thread.currentThread().getId() + " read < 0. and breaks, localRead=" + localRead);
						break;
					}
					loader.addFinishedSize(len);
					localFinished += len;
					randAccFile.write(buffer, 0, len);
				}
				inStream.close();
				randAccFile.close();
				con.disconnect();
				loader.latch.countDown();
//				System.out.println(Thread.currentThread().getId() + " count down. " + loader.latch.getCount() + "localFinished=" +localFinished);
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}
	}
	private static void setPosHeader(URLConnection con, long startPos, long endPos)
	{
		con.setRequestProperty("Range", "bytes=" + startPos + "-" + endPos);
	}
	@Override
	public void download(URL url, File saveFile) throws IOException
	{
		this.url = url;
		this.saveFile = saveFile;
		HttpURLConnection con = null;
		InputStream inStream = null;
		OutputStream outStream = null;
		con = (HttpURLConnection) url.openConnection();
		setHeader(con);
		// con.setConnectTimeout(10000);
		// con.setReadTimeout(10000);
		this.totalBytes = con.getContentLength();
		con.disconnect();
		ExecutorService threadPool = Executors.newFixedThreadPool(numOfThreads);
		latch = new CountDownLatch(numOfThreads);
		long cutSize = totalBytes / numOfThreads;
		for (int i = 0; i < numOfThreads; ++i)
		{
			long startPos = i * cutSize;
			long endPos = startPos + cutSize - 1;
			if (i == numOfThreads + 1)
				endPos = totalBytes;
			threadPool.execute(new ChildDownloadThread(this, startPos, endPos));
		}
		try
		{
			latch.await();
			threadPool.shutdown();
			System.out.println("All Threads finished download. SHUT DOWN");
		}
		catch (InterruptedException e)
		{
			e.printStackTrace();
		}
	}
	private static boolean setHeader(URLConnection con)
	{
		Properties prop = new Properties();
		try
		{
			prop.load(new FileInputStream("header.properties"));
		}
		catch (FileNotFoundException e)
		{
			e.printStackTrace();
			return false;
		}
		catch (IOException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		for (Iterator iter = prop.keySet().iterator(); iter.hasNext();)
		{
			String key = (String) iter.next();
			con.setRequestProperty(key, prop.getProperty(key));
		}
		return true;
	}
	@Override
	public long getFileSize()
	{
		return totalBytes;
	}
	@Override
	public double getFinishedRate()
	{
		return finishedBytes / (double) totalBytes;
	}
	@Override
	public long getFinishedSize()
	{
		return finishedBytes;
	}
	private synchronized void addFinishedSize(long size)
	{
		this.finishedBytes += size;
	}
	@Override
	public void pause()
	{
	}
	@Override
	public void resume()
	{
		
	}
	@Override
	public boolean isPasued()
	{
		return false;
	}
}

4.字符客户端代码：

public class NetLoaderCharClient
{
	/**
	 * @param args
	 * @throws MalformedURLException 
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws MalformedURLException, InterruptedException
	{
		final URL url = new URL("http://mdj.newhua.com/down/jpwb70yh.zip");
		final File f = new File("E://jpwb70yh.zip");
		DownloadService downService = new MultiThreadLoader(url, f, 8);
//		DownloadService downService = new SingleThreadLoader();
		Thread downThread = new Thread(new DownloadThread(url, f, downService));
		downThread.start();
		Thread showInfoThread = new Thread(new ShowDownInfoThread(downService, downThread)); 
		showInfoThread.start();
	}
	
	private static class DownloadThread implements Runnable
	{
		private URL url;
		private File f;
		private DownloadService downService;
		public DownloadThread(URL url, File f, DownloadService downService)
		{
			this.url = url;
			this.f = f;
			this.downService = downService;
		}
		public void run()
		{
			try
			{
				downService.download(url, f);
			}
			catch (Exception e)
			{
				e.printStackTrace();
				System.out.println("通信中断，IO异常");
			}
		}
	}
	
	private static class ShowDownInfoThread implements Runnable
	{
		private DownloadService downService;
		private Thread downThread;
		public ShowDownInfoThread(DownloadService downService, Thread downThread)
		{
			this.downService = downService;
			this.downThread = downThread;
		}
		public void run()
		{
			System.out.println("TotalSize=" + downService.getFileSize());
			System.out.println("Current finished Size     |Rate");
			int count = 0;
			while(downThread.isAlive())
			{
				System.out.printf("%10d/%10d     %.2f%%  %.2fKB/s%5d/n", downService.getFinishedSize(),  downService.getFileSize(), downService.getFinishedRate() * 100, (double)downService.getFinishedSize() / 1024  / count, ++count);
				try
				{
					Thread.sleep(1000);
				}
				catch (InterruptedException e)
				{
					e.printStackTrace();
				}
			}
			System.out.printf("%10d/%10d     %.2f%%  %.2fKB/s%5d/n", downService.getFinishedSize(),  downService.getFileSize(), downService.getFinishedRate() * 100, (double)downService.getFinishedSize() / 1024  / count, ++count);
			
		}
	}
}

多线程实现测试结果如下：

Current finished Size     |Rate
         0/         0     NaN%  NaNKB/s    1
    160654/   1788759     8.98%  156.89KB/s    2
    221310/   1788759     12.37%  108.06KB/s    3
    385386/   1788759     21.54%  125.45KB/s    4
    697238/   1788759     38.98%  170.22KB/s    5
    971982/   1788759     54.34%  189.84KB/s    6
   1072170/   1788759     59.94%  174.51KB/s    7
   1231890/   1788759     68.87%  171.86KB/s    8
   1348365/   1788759     75.38%  164.60KB/s    9
   1418377/   1788759     79.29%  153.90KB/s   10
   1598739/   1788759     89.38%  156.13KB/s   11
   1617929/   1788759     90.45%  143.64KB/s   12
   1674871/   1788759     93.63%  136.30KB/s   13
   1698103/   1788759     94.93%  127.56KB/s   14
   1714075/   1788759     95.82%  119.56KB/s   15
   1732951/   1788759     96.88%  112.82KB/s   16
   1763443/   1788759     98.58%  107.63KB/s   17
All Threads finished download. SHUT DOWN
   1788752/   1788759     100.00%  102.75KB/s   18

单线程实现测试结果如下：

   1649200/   1788759     92.20%  29.28KB/s   56
   1649200/   1788759     92.20%  28.76KB/s   57
   1705828/   1788759     95.36%  29.23KB/s   58
   1737772/   1788759     97.15%  29.26KB/s   59
   1769716/   1788759     98.94%  29.29KB/s   60
   1779880/   1788759     99.50%  28.97KB/s   61
   1788759/   1788759     100.00%  28.64KB/s   62

可以看到，使用8线程下载只用了18秒，最终平均速度为：107.75KB/s,而单线程下载却用了62秒，平均速度只有28.64KB/s，可见，多线程对下载的速度提升还是有很大帮助的。

而这里的多线程下载实现又有什么样的性能瓶颈呢？观察多线程下载的测试结果可以发现，刚开始速度非常快，等过了一段时间之后，速度明显有下降的趋势。这应该是因为，刚开始同时有多个线程下载，而到后来，某些线程下载完毕，只剩下极少数线程还在下载，并行降低，甚至退化成单线程下载，所以导致速度明显下降。

当然这里的实现可能还会存在诸多问题，并且断点续传还没有实现，希望最近能抽出一点时间来完成这个工作。