多執行緒下載原理解析

摘要： 先附上流程圖 1.入口DownLoadManager.download() /** * * @param request 請求實體引數Entity * @param tag 下載地址 * @param callBack 返回給呼叫的CollBack */ ...

先附上流程圖

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;
}
複製程式碼

單執行緒下載不需要偏移量

具體可以檢視 ofollow,noindex">github.com/lizubing199…