pieceloader.java

一个Mp3在线搜索器

/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package biz.tbuy.huliqing.jloading.downloader;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.URL;
import java.net.URLConnection;
import java.util.List;

/**
 *
 * @author huliqing
 */
public class PieceLoader extends Thread{
    private Downloader dl;      // 主下载管理
    private List<Piece> tasks;  // 任务列表
    
    private byte[] buff = new byte[1024 * 8];   // 缓冲区的大小
    private RandomAccessFile out;
    private BufferedInputStream in;
    
    private long readBytes; // 当前线程的读取字节数
    private long timeUsed;  // 当前线程在读取数据时总花费的时间
    
    public PieceLoader(Downloader dl, List<Piece> tasks) {
        this.dl = dl;
        this.tasks = tasks;
    }
    
    /** 获取当前线程已经下载的字节数 */
    public long getReadBytes() {
        return readBytes;
    }
    
    /** 获取当前线程读取数据时总花费的时间 */
    public long getTimeUsed() {
        return timeUsed;
    }
    
    /** 获取当前线程的下载速度 */
    public long getSpeed() {
        if (timeUsed <= 0) return 0;
        return readBytes / timeUsed;
    }
    
    /** 继续下载任务 */
    public synchronized void toContinue() {
        this.notifyAll();
    }
    
    @Override
    public void run() {
        while (!dl.isOk()) {
            
            // 暂停任务
            synchronized (this) {
                if (dl.isPaused()) {
                    try {
                        this.wait();
                    } catch (InterruptedException e) {
                    }
                }
            }
            
            // 中断停止
            if (Thread.interrupted() || dl.isStopped()) {
                return;
            }
            
            // 等待获取任务
            Piece piece;
            synchronized (tasks) {
                while (tasks.isEmpty()) {
                    if (dl.isOk()) return;
                    try {
                        tasks.wait();
                        //System.out.println(this.getName() + ":wait............");
                    } catch (InterruptedException ie) {
                        //System.out.println(this.getName() + 
                        //        ":InterruptedException:" + ie.getMessage());
                    }
                }
                piece = tasks.remove(0);
                dl.removeFreeLoader(this);
                //System.out.println(this.getName() + ":loading............");
            }
            try {
                URL u = new URL(dl.getURL());
                URLConnection uc = u.openConnection();
                // 设置断点续传位置
                uc.setAllowUserInteraction(true);
                uc.setRequestProperty("Range", "bytes=" + piece.getPos() + "-" + piece.getEnd());
                in = new BufferedInputStream(uc.getInputStream());

                out = new RandomAccessFile(dl.getFileProcess(), "rw");
                out.seek(piece.getPos()); // 设置指针位置

                long start;
                long end;
                int len = 0;
                while (piece.getPos() < piece.getEnd()) {
                    start = System.currentTimeMillis();
                    len = in.read(buff, 0, buff.length);
                    if (len == -1) break;
                    out.write(buff, 0, len);
                    end = System.currentTimeMillis();
                    timeUsed += end - start;    // 累计时间使用
                    
                    long newPos = piece.getPos() + len;
                    
                    // 如果该区段已经完成,如果该线程负责的区域已经完成，或出界
                    if (newPos > piece.getEnd()) {
                        piece.setPos(piece.getEnd());   
                        long offset = newPos - piece.getEnd();
                        long trueReads = (len - offset + 1);
                        dl.growReadBytes(trueReads);    // 修正偏移量
                        dl.setOffsetTotal(dl.getOffsetTotal() + trueReads);
                        readBytes += trueReads;
                        //System.out.println(this.getName() + ":read=" + trueReads);
                    } else {
                        dl.growReadBytes(len);
                        piece.setPos(piece.getPos() + len);
                        readBytes += len;
                        //System.out.println(this.getName() + ":read=" + len);
                    }
                    // 如果存在空闲的任务线程，则切割出新的区域至任务队列中。由空闲
                    // 的线程辅助下载
                    if (dl.isFreeLoader()) {
                        Piece newPiece = piece.cutPiece();
                        if (newPiece != null) {
                            synchronized (tasks) {
                                dl.addTask(newPiece);
                                dl.setRepairCount(dl.getRepairCount() + 1); // 增加切割次数
                                tasks.notifyAll();  // 唤醒等待任务中的空闲线程
                            }
                        }
                        
                    }
                    // 暂停任务
                    synchronized (this) {
                        if (dl.isPaused()) {
                            try {
                                this.wait();
                            } catch (InterruptedException e) {
                            }
                        }
                    }
                    
                    // 中断停止
                    if (Thread.interrupted() || dl.isStopped()) {
                        in.close();
                        out.close();
                        return;
                    }
                    //System.out.println(this.getName() + ":read:" + dl.getReadBytes());
                }
                out.close();
                in.close(); 
                dl.addFreeLoader(this);
                //System.out.println("切割次数：" + dl.getRepairCount());
                if (dl.isOk()) dl.processWhenOk();
            } catch (IOException e) {
                //System.out.println(this.getName() + ":无法读取数据");
            }
        }
    }
    
}