【移動應(yīng)用開發(fā)技術(shù)】Android多線程斷點續(xù)傳下載原理及實現(xiàn)

【移動應(yīng)用開發(fā)技術(shù)】Android多線程斷點續(xù)傳下載原理及實現(xiàn)

這段時間看了看工作室的工具庫的下載組件，發(fā)現(xiàn)其存在一些問題：

1.下載核心邏輯有bug，在暫停下載或下載失敗等情況時有概率無法順利完成下載。2.雖然原來的設(shè)計是采用多線程斷點續(xù)傳的設(shè)計，但打了一下日志發(fā)現(xiàn)其實下載任務(wù)都是在同一個線程下串行執(zhí)行，并沒有起到加快下載速度的作用。

1.下載核心邏輯有bug，在暫停下載或下載失敗等情況時有概率無法順利完成下載。2.雖然原來的設(shè)計是采用多線程斷點續(xù)傳的設(shè)計，但打了一下日志發(fā)現(xiàn)其實下載任務(wù)都是在同一個線程下串行執(zhí)行，并沒有起到加快下載速度的作用?？紤]到原來的代碼并不復(fù)雜，因此對這部分下載組件進(jìn)行了重寫。這里記錄一下里面的多線程斷點續(xù)傳功能的實現(xiàn)。請查看完整的PDF版(更多完整項目下載。未完待續(xù)。源碼。圖文知識后續(xù)上傳github。)可以點擊關(guān)于我聯(lián)系我獲取完整PDF(VX：m多線程下載意義首先我們談一談，多線程下載的意義。在日常的場景下，網(wǎng)絡(luò)中不可能只有下載方與服務(wù)器之間這樣一條連接，為了避免在這樣的場景下的網(wǎng)絡(luò)擁塞，TCP協(xié)議通過調(diào)節(jié)窗口的大小來避免出現(xiàn)擁塞，但這個窗口的大小可能沒辦法達(dá)到我們預(yù)期的效果：充分利用我們的帶寬。因此我們可以采用多個TCP連接的形式來提高我們帶寬的利用率，從而加快下載速度。打個比喻就是我們要從一個水缸中用抽水機(jī)通過水管抽水，由于管子的直徑等等的限制，我們單條管子無法完全利用我們的抽水機(jī)的抽水動力。因此我們就將這些抽水的任務(wù)分成了多份，分?jǐn)偟蕉鄠€管子上，這樣就可以更充分的利用我們的抽水機(jī)動力，從而提高抽水的速度。因此，我們使用多線程下載的主要意義就是——提高下載速度。多線程下載原理前面提到了我們主要的目的是將一個總的下載任務(wù)分?jǐn)偟蕉鄠€子任務(wù)中，比如假設(shè)我們用5個線程下載這個文件，那么我們就可以對一個長度為N的任務(wù)進(jìn)行如下圖的均分：但真實場景下往往N都不是剛好為5的倍數(shù)的，因此對于最后一個任務(wù)還需要加上剩余的任務(wù)量，也就是N/5+N%5。HttpRange請求頭上面的任務(wù)分配我們已經(jīng)了解了，看起來很理想，但有一個問題，我們?nèi)绾螌崿F(xiàn)向服務(wù)器只請求這個文件的某一段而不是全部呢？我們可以通過在請求頭中加入Range字段來指定請求的范圍，從而實現(xiàn)指定某一段的數(shù)據(jù)。如：RANGEbytes=10000-19999就指定了10000-19999這段字節(jié)的數(shù)據(jù)所以我們的核心思想就是通過它拿到文件對應(yīng)字節(jié)段的InputStream，然后對它讀取并寫入文件。RandomAccessFile文件寫入下面再講講文件寫入問題，由于我們是多線程下載，因此文件并不是每次都是從前往后一個個字節(jié)寫入的，隨時可能在文件的任何一個地方寫入數(shù)據(jù)。因此我們需要能夠在文件的指定位置寫入數(shù)據(jù)。這里我們用到了RandomAccessFile來實現(xiàn)這個功能。RandomAccessFile是一個隨機(jī)訪問文件類，同時整合了FileOutputStream和FileInputStream，支持從文件的任何字節(jié)處讀寫數(shù)據(jù)。通過它我們就可以在文件的任何字節(jié)處寫入數(shù)據(jù)。接下來簡單講講我們這里是如何使用RandomAccessFile的。我們對于每個子任務(wù)來說都有一個開始和結(jié)束的位置。每個任務(wù)都可以通過RandomAccessFile::seek跳轉(zhuǎn)到文件的對應(yīng)字節(jié)位置，然后從該位置開始讀取InputStream并寫入。這樣，就實現(xiàn)了不同線程對文件的隨機(jī)寫入。文件大小的獲取由于我們在真正開始下載之前，我們需要先將任務(wù)分配到各個線程，因此我們需要先了解到文件的大小。為了獲取到文件的大小，我們用到ResponseHeaders中的Content-Length字段。如下圖所示，可以看到，打開該下載請求的鏈接后，ResponseHeaders中包含了我們需要的Content-Length，也就是該文件的大小，單位是字節(jié)。斷點續(xù)傳原理對于多個子任務(wù)，我們?nèi)绾螌崿F(xiàn)它們的斷點續(xù)傳呢？其實原理很簡單，只需要保證每個子任務(wù)的下載進(jìn)度能夠被即時地記錄即可。這樣繼續(xù)下載時只需要讀取這些下載記錄，從上次下載結(jié)束的位置開始下載即可。它的實現(xiàn)有很多方式，只要能做到數(shù)據(jù)持久化即可。這里我使用的是數(shù)據(jù)庫來實現(xiàn)。這樣，我們的子任務(wù)需要擁有一些必要的信息通過這些信息，我們就能夠記錄子任務(wù)的下載進(jìn)度從而恢復(fù)我們之前的下載，實現(xiàn)斷點續(xù)傳。代碼實現(xiàn)下面我們用代碼來實現(xiàn)這樣一個多線程下載功能。下載狀態(tài)首先，我們定義一下下載中的各個狀態(tài)：publicclassDownloadStatus{

publicstaticfinalintIDLE=233;//空閑，默認(rèn)狀態(tài)

publicstaticfinalintCOMPLETED=234;//完成

publicstaticfinalintDOWNLOADING=235;//下載中

publicstaticfinalintPAUSE=236;//暫停

publicstaticfinalintERROR=237;//出錯

}可以看到，這里定義了如上的五種狀態(tài)。這里需要用到如數(shù)據(jù)庫及HTTP請求的功能，我們這里定義其接口如下，具體實現(xiàn)各位可以根據(jù)需要自己實現(xiàn)：publicinterfaceDownloadDbHelper{

/**

*從數(shù)據(jù)庫中刪除子任務(wù)記錄

*@paramtask子任務(wù)記錄

*/

voiddelete(SubDownloadTasktask);

/**

*向數(shù)據(jù)庫中插入子任務(wù)記錄

*@paramtask子任務(wù)記錄

*/

voidinsert(SubDownloadTasktask);

/**

*在數(shù)據(jù)庫中更新子任務(wù)記錄

*@paramtask子任務(wù)記錄

*/

voidupdate(SubDownloadTasktask);

/**

*獲取所有指定Task下的子任務(wù)記錄

*@paramtaskTagTask的Tag

*@return子任務(wù)記錄

*/

List<SubDownloadTask>queryByTaskTag(StringtaskTag);

}publicinterfaceDownloadHttpHelper{

/**

*獲取文件總長度

*@paramurl下載url

*@paramcallback獲取文件長度CallBack

*/

voidgetTotalSize(Stringurl,NetCallback<Long>callback);

/**

*獲取InputStream

*@paramurl下載url

*@paramstart開始位置

*@paramend結(jié)束位置

*@paramcallback獲取字節(jié)流的CallBack

*/

voidgetStreamByRange(Stringurl,longstart,longend,NetCallback<InputStream>callback);

}我們先從上到下，從子任務(wù)開始實現(xiàn)。在我的設(shè)計中，它具有如下的成員變量：@Entity

publicclassSubDownloadTaskimplementsRunnable{

publicstaticfinalintBUFFER_SIZE=1024*1024;

privatestaticfinalStringTAG=SubDownloadTask.class.getSimpleName();

@Id

privateLongid;

privateStringurl;//文件下載的url

privateStringtaskTag;//父任務(wù)的Tag

privatelongtaskSize;//子任務(wù)大小

privatelongcompletedSize;//子任務(wù)完成大小

privatelongstartPos;//開始位置

privatelongcurrentPos;//當(dāng)前位置

privatelongendPos;//結(jié)束位置

privatevolatileintstatus;//當(dāng)前下載狀態(tài)

@Transient

privateSubDownloadListenerlistener;//子任務(wù)下載監(jiān)聽，主要用于提示父任務(wù)

@Transient

privateFilesaveFile;//要保存到的文件

...

}由于這里的數(shù)據(jù)庫的操作是用GreenDao實現(xiàn)，因此這里有一些相關(guān)注解，各位可以忽略。可以看到，子任務(wù)是一個Runnable，我們可以通過其run方法開始下載，這樣就可以通過如ExecutorService來開啟多個線程執(zhí)行子任務(wù)。我們看到其run方法：@Override

publicvoidrun(){

status=DownloadStatus.DOWNLOADING;

DownloadManager.getInstance()

.getHttpHelper()

.getStreamByRange(url,currentPos,endPos,newNetCallback<InputStream>(){

@Override

publicvoidonResult(InputStreaminputStream){

listener.onSubStart();

writeFile(inputStream);

}

@Override

publicvoidonError(Stringmessage){

listener.onSubError("文件流獲取失敗");

status=DownloadStatus.ERROR;

}

});

}可以看到，我們獲取了其從currentPos到endPos端的字節(jié)流，通過其ResponseBody拿到了它的InputStream，然后調(diào)用了writeFile(InputStream)方法進(jìn)行文件的寫入。文件寫入接下來看到writeFile方法：privatevoidwriteFile(InputStreamin){

try{

RandomAccessFilefile=newRandomAccessFile(saveFile,"rwd");//通過saveFile建立RandomAccessFile

file.seek(currentPos);//跳轉(zhuǎn)到對應(yīng)位置

byte[]buffer=newbyte[BUFFER_SIZE];

while(true){

//循環(huán)讀取InputStream，直到暫?；蜃x取結(jié)束

if(status!=DownloadStatus.DOWNLOADING){

//狀態(tài)不為DOWNLOADING，停止下載

break;

}

intoffset=in.read(buffer,0,BUFFER_SIZE);

if(offset==-1){

//讀取不到數(shù)據(jù)，說明讀取結(jié)束

break;

}

//將讀取到的數(shù)據(jù)寫入文件

file.write(buffer,0,offset);

//下載數(shù)據(jù)并在數(shù)據(jù)庫中更新

currentPos+=offset;

completedSize+=offset;

DownloadManager.getInstance()

.getDbHelper()

.update(this);

//通知父任務(wù)下載進(jìn)度

listener.onSubDownloading(offset);

}

if(status==DownloadStatus.DOWNLOADING){

//下載完成

status=DownloadStatus.COMPLETED;

//通知父任務(wù)下載完成

listener.onSubComplete(completedSize);

}

file.close();

in.close();

}catch(IOExceptione){

e.printStackTrace();

listener.onSubError("文件下載失敗");

status=DownloadStatus.ERROR;

resetTask();

}

}具體流程可以看代碼中的注釋。可以看到，子任務(wù)實際上就是循環(huán)讀取InputStream，并寫入文件，同時將下載進(jìn)度同步到數(shù)據(jù)庫。父任務(wù)也就是我們具體的下載任務(wù)，我們同樣先看到成員變量：publicclassDownloadTaskimplementsSubDownloadListener{

privatestaticfinalStringTAG=DownloadTask.class.getSimpleName();

privateStringtag;//下載任務(wù)的Tag，用于區(qū)分不同下載任務(wù)

privateStringurl;//下載url

privateStringsavePath;//保存路徑

privateStringfileName;//保存文件名

privateDownloadListenerlistener;//下載監(jiān)聽

privatelongcompleteSize;//下載完成大小

privatelongtotalSize;//下載任務(wù)總大小

privateintstatus;//當(dāng)前下載進(jìn)度

privateintthreadNum;//線程數(shù)（由外部設(shè)置的每個任務(wù)的下載線程數(shù)）

privateFilefile;//保存文件

privateList<SubDownloadTask>subTasks;//子任務(wù)列表

privateExecutorServicemExecutorService;//線程池，用于執(zhí)行子任務(wù)

...

}對于一個下載任務(wù)，可以通過download方法開始執(zhí)行：publicvoiddownload(){

listener.onStart();

subTasks=querySubTasks();

status=DownloadStatus.DOWNLOADING;

if(subTasks.isEmpty()){

//是新任務(wù)

downloadNewTask();

}elseif(subTasks.size()==threadNum){

//不是新任務(wù)

downloadExistTask();

}else{

//不是新任務(wù)，但下載線程數(shù)有誤

listener.onError("斷點數(shù)據(jù)有誤");

resetTask();

}

}可以看到，我們先將子任務(wù)列表從數(shù)據(jù)庫中讀取出來。如果子任務(wù)列表大小不等于線程數(shù)，說明當(dāng)前的下載記錄已不可用，于是重置下載任務(wù)，從新下載。我們先看到downloadNewTask方法：可以看到，獲取到總長度后，通過調(diào)用initSubTasks方法，對子任務(wù)列表進(jìn)行了初始化（計算子任務(wù)長度等），然后調(diào)用了startAsyncDownload方法后通過ExecutorService運行子任務(wù)進(jìn)入子任務(wù)進(jìn)行下載。我們看到initSubTasks方法：privatevoidinitSubTasks(){

longaverageSize=totalSize/threadNum;

for(inttaskIndex=0;taskIndex<threadNum;taskIndex++){

longtaskSize=averageSize;

if(taskIndex==threadNum-1){

//最后一個任務(wù)，則size還需要加入剩余量

taskSize+=totalSize%threadNum;

}

longstart=0L;

intindex=taskIndex;

while(index>0){

start+=subTasks.get(index-1).getTaskSize();

index--;

}

longend=start+taskSize-1;//注意這里

SubDownloadTasksubTask=newSubDownloadTask();

subTask.setUrl(url);

subTask.setStatus(DownloadStatus.IDLE);

subTask.setTaskTag(tag);

subTask.setCompletedSize(0);

subTask.setTaskSize(taskSize);

subTask.setStartPos(start);

subTask.setCurrentPos(start);

subTask.setEndPos(end);

subTask.setSaveFile(file);

subTask.setListener(this);

DownloadManager.getInstance()

.getDbHelper()

.insert(subTask);

subTasks.add(subTask);

}

}可以看到就是計算每個任務(wù)的大小及開始及結(jié)束點的位置，這里要注意的是endPos需要-1，否則各個任務(wù)的下載位置會重疊，并且最后一個任務(wù)會多下載一個字節(jié)導(dǎo)致如文件損壞等影響。具體原因就是比如一個大小為500的文件，則應(yīng)當(dāng)是0-499而不是0-500。接下來我們看看downloadExistTask方法：privatevoiddownloadExistTask(){

//不是新任務(wù)，且下載線程數(shù)無誤，計算已下載大小

completeSize=countCompleteSize();

totalSize=countTotalSize();

startAsyncDownload();

}這里其實很簡單，遍歷子任務(wù)列表計算已下載量及總?cè)蝿?wù)量，并調(diào)用startAsyncDownload開始多線程下載。具體執(zhí)行子任務(wù)我們可以看到startAsyncDownload方法：privatevoidstartAsyncDownload(){

for(SubDownloadTasksubTask:subTasks){

if(subTask.getCompletedSize()<subTask.getTaskSize()){

//只下載沒有下載結(jié)束的子任務(wù)

mExecutorService.execute(subTask);

}

}

}可以看到，這里其實只是通過ExecutorService執(zhí)行對應(yīng)子任務(wù)（Runnable）而已。####暫停功能我們接下來看到pause方法：publicvoidpause(){

stopAsyncDownload();

status=DownloadStatus.PAUSE;

listener.onPause();

}可以看到，這里只是調(diào)用了stopAsyncDownload方法停止子任務(wù)?？吹絪topAsyncDownload方法：privatevoidstopAsyncDownload(){

for(SubDownloadTasksubTask:subTasks){

if(subTask.getStatus()!=DownloadStatus.COMPLETED){

//下載完成的不再取消

subTask.cancel();

}

}

}可以看到，調(diào)用了子任務(wù)的cancel方法。繼續(xù)看到子任務(wù)的cancel方法：voidcancel(){

status=DownloadStatus.PAUSE;

listener.onSubCancel();

}這里很簡單，僅僅是將下載狀態(tài)設(shè)置為了PAUSE，這樣在寫入文件的下一次while循環(huán)時便會中止循環(huán)從而結(jié)束Runnable的執(zhí)行。看到cancel方法：publicvoidcancel(){

stopAsyncDownload();

resetTask();

listener.onCancel();

}可以看到和暫停的邏輯差不多，只是在暫停后還需要對子任務(wù)重置從而使得下次下載從頭開始。底層到上層的通知機(jī)制前面提到，外部可以通過DownloadListener監(jiān)聽下載的進(jìn)度，下面是DownloadListener接口的定義：publicinterfaceDownloadListener{

defaultvoidonStart(){}

defaultvoidonDownloading(longprogress,longtotal){}

defaultvoidonPause(){}

defaultvoidonCancel(){}

defaultvoidonComplete(){}

defaultvoidonError(Stringmessage){}

}我們實時的下載進(jìn)度其實是在子任務(wù)的保存文件過程中才能體現(xiàn)出來的，同樣，子任務(wù)的下載失敗也需要通知到DownloadListener，這是怎么做到的呢？前面提到了，我們還定義了一個SubDownloadListener，其監(jiān)聽者就是子任務(wù)的父任務(wù)。通過監(jiān)聽我們可以將子任務(wù)狀態(tài)反饋到父任務(wù)，父任務(wù)再根據(jù)具體情況反饋數(shù)據(jù)給Down