阅读 526

多线程下载原理解析

先附上流程图

1.入口DownLoadManager.download()

/**
 *
 * @param request 请求实体参数Entity
 * @param tag 下载地址
 * @param callBack 返回给调用的CollBack
 */
public void download(DownloadRequest request, String tag, CallBack callBack) {
    final String key = createKey(tag);
    if (check(key)) {
        // 请求的响应 需要状态传递类 以及对应的回调
        DownloadResponse response = new DownloadResponseImpl(mDelivery, callBack);
        // 下载器 需要线程池 数据库管理者 对应的url key值 之后回调给自己
        Downloader downloader = new DownloaderImpl(request, response,
            mExecutorService, mDBManager, key, mConfig, this);
        mDownloaderMap.put(key, downloader);
        //开始下载
        downloader.start();
    }
}
复制代码

DownloadResponseImpl 下载响应需要把本身的下载事件回调给调用者,由于下载是在子线程里面的,所以专门搞了一个下载状态的传递类

DownLoaderImpl 下载器 需要的参数就比较多了,请求实体,对应的下载响应,线程池,数据库管理器,url的hash值,对应的配置,还有下载的回调

加入进LinkedHashMap 做一个有序的存储

之后调用下载器的start方法

2.开始下载 start

  @Override
  public void start() {
    //修改为Started状态
    mStatus = DownloadStatus.STATUS_STARTED;
    //CallBack 回调给调用者
    mResponse.onStarted();
    // 连接获取是否支持多线程下载
    connect();
    }
/**
 * 执行连接任务
 */
private void connect() {
    mConnectTask = new ConnectTaskImpl(mRequest.getUri(), this);
    mExecutor.execute(mConnectTask);
}
复制代码

在正式下载之前需要确定后台是否支持断点下载,所以才有先执行这个ConnectTaskImpl 连接任务

3.ConnectTaskImpl 连接任务

  @Override
  public void run() {
    // 设置为后台线程
    Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
    //修改连接中状态
    mStatus = DownloadStatus.STATUS_CONNECTING;
    //回调给调用者
    mOnConnectListener.onConnecting();
    try {
      //执行连接方法
      executeConnection();
    } catch (DownloadException e) {
      handleDownloadException(e);
    }
  }

  /**
   *
   * @throws DownloadException
   */
  private void executeConnection() throws DownloadException {
    mStartTime = System.currentTimeMillis();
    HttpURLConnection httpConnection = null;
    final URL url;
    try {
      url = new URL(mUri);
    } catch (MalformedURLException e) {
      throw new DownloadException(DownloadStatus.STATUS_FAILED, "Bad url.", e);
    }
    try {
      httpConnection = (HttpURLConnection) url.openConnection();
      httpConnection.setConnectTimeout(Constants.HTTP.CONNECT_TIME_OUT);
      httpConnection.setReadTimeout(Constants.HTTP.READ_TIME_OUT);
      httpConnection.setRequestMethod(Constants.HTTP.GET);
      httpConnection.setRequestProperty("Range", "bytes=" + 0 + "-");
      final int responseCode = httpConnection.getResponseCode();
      if (responseCode == HttpURLConnection.HTTP_OK) {
        //后台不支持断点下载,启用单线程下载
        parseResponse(httpConnection, false);
      } else if (responseCode == HttpURLConnection.HTTP_PARTIAL) {
        //后台支持断点下载,启用多线程下载
        parseResponse(httpConnection, true);
      } else {
        throw new DownloadException(DownloadStatus.STATUS_FAILED,
            "UnSupported response code:" + responseCode);
      }
    } catch (ProtocolException e) {
      throw new DownloadException(DownloadStatus.STATUS_FAILED, "Protocol error", e);
    } catch (IOException e) {
      throw new DownloadException(DownloadStatus.STATUS_FAILED, "IO error", e);
    } finally {
      if (httpConnection != null) {
        httpConnection.disconnect();
      }
    }
  }

  private void parseResponse(HttpURLConnection httpConnection, boolean isAcceptRanges)
      throws DownloadException {

    final long length;
    //header获取length
    String contentLength = httpConnection.getHeaderField("Content-Length");
    if (TextUtils.isEmpty(contentLength) || contentLength.equals("0") || contentLength
        .equals("-1")) {
      //判断后台给你length,为null 0,-1,从连接中获取
      length = httpConnection.getContentLength();
    } else {
      //直接转化
      length = Long.parseLong(contentLength);
    }

    if (length <= 0) {
      //抛出异常数据
      throw new DownloadException(DownloadStatus.STATUS_FAILED, "length <= 0");
    }
    //判断是否取消和暂停
    checkCanceledOrPaused();

    //Successful
    mStatus = DownloadStatus.STATUS_CONNECTED;
    //获取时间差
    final long timeDelta = System.currentTimeMillis() - mStartTime;
    //回调给调用者
    mOnConnectListener.onConnected(timeDelta, length, isAcceptRanges);
  }

  private void checkCanceledOrPaused() throws DownloadException {
    if (isCanceled()) {
      // cancel
      throw new DownloadException(DownloadStatus.STATUS_CANCELED, "Connection Canceled!");
    } else if (isPaused()) {
      // paused
      throw new DownloadException(DownloadStatus.STATUS_PAUSED, "Connection Paused!");
    }
  }

  //统一执行对应的异常信息
  private void handleDownloadException(DownloadException e) {
    switch (e.getErrorCode()) {
      case DownloadStatus.STATUS_FAILED:
        synchronized (mOnConnectListener) {
          mStatus = DownloadStatus.STATUS_FAILED;
          mOnConnectListener.onConnectFailed(e);
        }
        break;
      case DownloadStatus.STATUS_PAUSED:
        synchronized (mOnConnectListener) {
          mStatus = DownloadStatus.STATUS_PAUSED;
          mOnConnectListener.onConnectPaused();
        }
        break;
      case DownloadStatus.STATUS_CANCELED:
        synchronized (mOnConnectListener) {
          mStatus = DownloadStatus.STATUS_CANCELED;
          mOnConnectListener.onConnectCanceled();
        }
        break;
      default:
        throw new IllegalArgumentException("Unknown state");
    }
  }
复制代码

HttpURLConnection.HTTP_OK 不支持断点下载 使用单线程下载

HttpURLConnection.HTTP_PARTIAL 支持断点下载 使用多线程下载

如果成功就会回调到OnConnectListener.onConnected(timeDelta, length, isAcceptRanges)方法中

回到下载器查看onConnected方法

4.查看下载器的onConnected()

@Override
public void onConnected(long time, long length, boolean isAcceptRanges) {
    if (mConnectTask.isCanceled()) {
        //连接取消
        onConnectCanceled();
    } else {
        mStatus = DownloadStatus.STATUS_CONNECTED;
        //回调给你响应连接成功状态
        mResponse.onConnected(time, length, isAcceptRanges);
        mDownloadInfo.setAcceptRanges(isAcceptRanges);
        mDownloadInfo.setLength(length);
        //真正开始下载
        download(length, isAcceptRanges);
    }
}

@Override
public void onConnectCanceled() {
    deleteFromDB();
    deleteFile();
    mStatus = DownloadStatus.STATUS_CANCELED;
    mResponse.onConnectCanceled();
    onDestroy();
}

@Override
public void onDestroy() {
    // trigger the onDestroy callback tell download manager
    mListener.onDestroyed(mTag, this);
}
复制代码

根据状态来处理,isCanceled() 删除数据库里面的数据,删除文件,更改为取消状态状态

未取消,进去下载download

5.下载文件download方法

/**
 * 下载开始
 * @param length 设置下载的长度
 * @param acceptRanges 是否支持断点下载
 */
private void download(long length, boolean acceptRanges) {
    mStatus = DownloadStatus.STATUS_PROGRESS;
    initDownloadTasks(length, acceptRanges);
    //开始下载任务
    for (DownloadTask downloadTask : mDownloadTasks) {
        mExecutor.execute(downloadTask);
    }
}

/**
 * 初始化下载任务
 * @param length
 * @param acceptRanges
 */
private void initDownloadTasks(long length, boolean acceptRanges) {
    mDownloadTasks.clear();
    if (acceptRanges) {
        List<ThreadInfo> threadInfos = getMultiThreadInfos(length);
        // init finished
        int finished = 0;
        for (ThreadInfo threadInfo : threadInfos) {
            finished += threadInfo.getFinished();
        }
        mDownloadInfo.setFinished(finished);
        for (ThreadInfo info : threadInfos) {
            //开始多线程下载
            mDownloadTasks.add(new MultiDownloadTask(mDownloadInfo, info, mDBManager, this));
        }
    } else {
        //单线程下载不需要保存进度信息
        ThreadInfo info = getSingleThreadInfo();
        mDownloadTasks.add(new SingleDownloadTask(mDownloadInfo, info, this));
    }
}

//TODO
private List<ThreadInfo> getMultiThreadInfos(long length) {
    // init threadInfo from db
    final List<ThreadInfo> threadInfos = mDBManager.getThreadInfos(mTag);
    if (threadInfos.isEmpty()) {
        final int threadNum = mConfig.getThreadNum();
        for (int i = 0; i < threadNum; i++) {
            // calculate average
            final long average = length / threadNum;
            final long start = average * i;
            final long end;
            if (i == threadNum - 1) {
                end = length;
            } else {
                end = start + average - 1;
            }
            ThreadInfo threadInfo = new ThreadInfo(i, mTag, mRequest.getUri(), start, end, 0);
            threadInfos.add(threadInfo);
        }
    }
    return threadInfos;
}

//单线程数据
private ThreadInfo getSingleThreadInfo() {
    ThreadInfo threadInfo = new ThreadInfo(0, mTag, mRequest.getUri(), 0);
    return threadInfo;
}
复制代码

根据connected返回的数据判断是否支持断点下载,支持acceptRanges 就调用getMultiThreadInfos来组装多线程下载数据,多线程需要初始化下载的进度信息,二单线程getSingleThreadInfo自己组装一个简单的就可以了

6.执行DownloadTaskImpl

@Override
public void run() {
	Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
	// 插入数据库
	insertIntoDB(mThreadInfo);
	try {
	  mStatus = DownloadStatus.STATUS_PROGRESS;
	  executeDownload();
	  //根据回调对象,加锁
	  synchronized (mOnDownloadListener) {
	    //没出异常就代表下载完成了
	    mStatus = DownloadStatus.STATUS_COMPLETED;
	    mOnDownloadListener.onDownloadCompleted();
	  }
	} catch (DownloadException e) {
	  handleDownloadException(e);
	}
}

	/**
   * 开始下载数据
   */
  	private void executeDownload() throws DownloadException {
    final URL url;
    try {
      url = new URL(mThreadInfo.getUri());
    } catch (MalformedURLException e) {
      throw new DownloadException(DownloadStatus.STATUS_FAILED, "Bad url.", e);
    }

    HttpURLConnection httpConnection = null;
    try {
      //设置http连接信息
      httpConnection = (HttpURLConnection) url.openConnection();
      httpConnection.setConnectTimeout(HTTP.CONNECT_TIME_OUT);
      httpConnection.setReadTimeout(HTTP.READ_TIME_OUT);
      httpConnection.setRequestMethod(HTTP.GET);
      //设置header数据,断点下载设置关键
      setHttpHeader(getHttpHeaders(mThreadInfo), httpConnection);
      final int responseCode = httpConnection.getResponseCode();
      if (responseCode == getResponseCode()) {
        //下载数据
        transferData(httpConnection);
      } else {
        throw new DownloadException(DownloadStatus.STATUS_FAILED,
            "UnSupported response code:" + responseCode);
      }
    } catch (ProtocolException e) {
      throw new DownloadException(DownloadStatus.STATUS_FAILED, "Protocol error", e);
    } catch (IOException e) {
      throw new DownloadException(DownloadStatus.STATUS_FAILED, "IO error", e);
    } finally {
      if (httpConnection != null) {
        httpConnection.disconnect();
      }
    }
  }

  /**
   * 设置header数据
   *
   * @param headers header元数据
   */
  private void setHttpHeader(Map<String, String> headers, URLConnection connection) {
    if (headers != null) {
      for (String key : headers.keySet()) {
        connection.setRequestProperty(key, headers.get(key));
      }
    }
  }

  /**
   * 下载数据
   */
  private void transferData(HttpURLConnection httpConnection) throws DownloadException {
    InputStream inputStream = null;
    RandomAccessFile raf = null;
    try {
      try {
        inputStream = httpConnection.getInputStream();
      } catch (IOException e) {
        throw new DownloadException(DownloadStatus.STATUS_FAILED, "http get inputStream error", e);
      }
	  //获取下载的偏移量
      final long offset = mThreadInfo.getStart() + mThreadInfo.getFinished();
      try {
		//设置偏移量
        raf = getFile(mDownloadInfo.getDir(), mDownloadInfo.getName(), offset);
      } catch (IOException e) {
        throw new DownloadException(DownloadStatus.STATUS_FAILED, "File error", e);
      }
      //开始写入数据
      transferData(inputStream, raf);
    } finally {
      try {
        IOCloseUtils.close(inputStream);
        IOCloseUtils.close(raf);
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }

  /**
   * 写入数据
   */
  private void transferData(InputStream inputStream, RandomAccessFile raf)
      throws DownloadException {
    final byte[] buffer = new byte[1024 * 8];
    while (true) {
      checkPausedOrCanceled();
      int len = -1;
      try {
        len = inputStream.read(buffer);
        if (len == -1) {
          break;
        }
        raf.write(buffer, 0, len);
        //设置下载的信息
        mThreadInfo.setFinished(mThreadInfo.getFinished() + len);
        synchronized (mOnDownloadListener) {
          mDownloadInfo.setFinished(mDownloadInfo.getFinished() + len);
          //回调进度
          mOnDownloadListener
              .onDownloadProgress(mDownloadInfo.getFinished(), mDownloadInfo.getLength());
        }
      } catch (IOException e) {
        //更新数据库
        updateDB(mThreadInfo);
        throw new DownloadException(DownloadStatus.STATUS_FAILED, e);
      }
    }
  }
复制代码

断点下载的关键是在header头信息里面添加已经下载length ,下载数据也是从下载的length点开始写入数据,写入数据,每个线程在对应的片段里面下载对应的数据,后期使用RandomAccessFile组装起来,合成一个文件

7.多线程下载时的header数据,以及RandomAccessFile组装

  @Override
  protected Map<String, String> getHttpHeaders(ThreadInfo info) {
    Map<String, String> headers = new HashMap<String, String>();
    //计算开始和结束的位置
    long start = info.getStart() + info.getFinished();
    long end = info.getEnd();
    headers.put("Range", "bytes=" + start + "-" + end);
    return headers;
  }

  @Override
  protected RandomAccessFile getFile(File dir, String name, long offset) throws IOException {
    File file = new File(dir, name);
    RandomAccessFile raf = new RandomAccessFile(file, "rwd");
    //设置偏移量
    raf.seek(offset);
    return raf;
  }
复制代码

8.单线程下载

  @Override
  protected Map<String, String> getHttpHeaders(ThreadInfo info) {
    // simply return null
    return null;
  }

  @Override
  protected RandomAccessFile getFile(File dir, String name, long offset) throws IOException {
    File file = new File(dir, name);
    RandomAccessFile raf = new RandomAccessFile(file, "rwd");
    raf.seek(0);
    return raf;
  }
复制代码

单线程下载不需要偏移量

具体可以查看github.com/lizubing199…

关注下面的标签,发现更多相似文章
评论