subtaskthread.java

一个java编写的下载器源码

package com.leapget.util;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.RandomAccessFile;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.Observable;

/**
 * 实施任务下载的子线程
 * 
 * @author  0501xql
 * @date    2008-8-4
 * @version 1.0
 */
public class SubTaskThread extends Observable implements Runnable {

	private TaskThread taskThread;	// 当前线程所属的主线程
	private int startPos;			// 开始位置
	private int endPos;				// 结束位置
	private int part;				// 当前下载文件的第几部分
	private boolean finished;		// 标识各线程是否下载完毕
	private byte[] buf;				// 缓冲大小
	private int delay = 10;			// 默认每隔10毫秒调用一次
	private boolean downloading = true;

	public SubTaskThread(TaskThread taskThread, int startPos, int endPos, int part) {
		this.taskThread = taskThread;
		this.startPos = startPos;
		this.endPos = endPos;
		this.part = part;
		this.finished = false;
		this.buf = new byte[1024*8];// 默认缓冲8KB
	}

	public int getPart() {
		return part;
	}

	public void setPart(int part) {
		this.part = part;
	}

	public boolean isFinished() {
		return finished;
	}

	public void setFinished(boolean finished) {
		this.finished = finished;
	}

	public TaskThread getTaskThread() {
		return taskThread;
	}
	
	public TaskInfo getTask(){
		return taskThread.getTask();
	}

	public void run() {
		URL downUrl = null;
		HttpURLConnection conn = null;
		BufferedInputStream bis = null;
		RandomAccessFile raf = null;
		TaskInfo task = taskThread.getTask();
		File tmpFile = taskThread.getTmpFile();						// 该线程生成的临时文件
		int len = -1;
		try {// 开始下载
			downUrl = new URL(task.getFileUrl());
			conn = (HttpURLConnection) downUrl.openConnection();
			conn.setRequestProperty("Range","bytes=" + startPos + "-");//设置已下载长度,注意要在连接建立以前设置
			if (conn.getResponseCode()/100 != 2 					//http响应代码2xx 是正常状态
				|| conn.getContentLength()<1){
					error();
					return;
			}
			bis = new BufferedInputStream(conn.getInputStream());
			raf = new RandomAccessFile(tmpFile,"rw");				// 随机读取写入文件
			raf.seek(startPos);										// 跳过前面的内容

			while (task.getState() == TaskInfo.TASK_NORMAL 
					&& downloading
					&& (-1 != (len = bis.read(buf)))) {
				if (Debug.isDebug) {
					System.out.println(tmpFile.getName() + " 线程 " + part
							+ " 读取从 " + (startPos + 1) + " 到 " + endPos
							+ " 部分的数据，实际读取 " + len + " 字节。");
				}
				if ((startPos + len) >= endPos) {
					len = endPos - startPos;
					downloading = false;				// 该部分下载完成
				}
				raf.write(buf, 0, len);
				startPos += len;
				task.addDownloadedBytes(len);
				dataChanged("downloading");				// 通知观察者数据已经改变
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (bis != null) bis.close();
				if (raf != null) raf.close();
				if (conn != null) conn.disconnect();	// 关闭连接
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		if(task.getState() == TaskInfo.TASK_NORMAL){	// 正常执行完下载
			if (Debug.isDebug) {
				System.out.println(tmpFile.getName() + " 线程 " + part + " 下载完毕！");
			}
	
			this.finished = true;							// 当前线程下载完毕
			boolean taskFinished = true;
			for(SubTaskThread s : taskThread.getSubThreads()){
				if(!s.isFinished()){
					taskFinished = false;
					break;
				}
			}
			if(taskFinished) dataChanged("taskfinished");
			else dataChanged("subthreadfinished");			// 通知观察者当前线程完毕
		}else{
			if (Debug.isDebug) {
				System.out.println(tmpFile.getName() + " 线程 " + part + " 因为 " + task.getStrState() + " 而停止.");
			}
		}
	}
	private void error(){
		System.out.println("********* 子线程下载错误！ *********");
	}
	private void dataChanged(Object arg) {
		setChanged();			// 设置一个内部标志位注明数据发生了变化
		notifyObservers(arg);	// 调用所有的观察者Observer的update()方法，通知它们数据发生了变化。
	}
}