RxDownload2 源碼解析(二)
源碼解析，如需轉載，請註明做者：Yuloran (t.cn/EGU6c76)java
前言
造輪子者：Season_zlc併發
本文主要講述 RxDownload2 的線程調度異步
下載任務分發線程
顧名思義，就是分發下載任務的線程。該線程運行在 DownloadService 中，從業務上看，DownloadService應當僅被 start() & bind() 一次。任務分發線程，在 onBind() 時建立：ide
- start & bind service [-> RxDownload.java]
/** * start and bind service. * * @param callback Called when service connected. */
private void startBindServiceAndDo(final ServiceConnectedCallback callback) {
Intent intent = new Intent(context, DownloadService.class);
intent.putExtra(DownloadService.INTENT_KEY, maxDownloadNumber);
context.startService(intent);
context.bindService(intent, new ServiceConnection() {
@Override
public void onServiceConnected(ComponentName name, IBinder binder) {
DownloadService.DownloadBinder downloadBinder
= (DownloadService.DownloadBinder) binder;
downloadService = downloadBinder.getService();
context.unbindService(this);
bound = true;
callback.call();
}
@Override
public void onServiceDisconnected(ComponentName name) {
//注意!!這個方法只會在系統殺掉Service時纔會調用!!
bound = false;
}
}, Context.BIND_AUTO_CREATE);
}
複製代碼
上述代碼有個細節，onServiceConnected() 中立刻調了 unbindService()。post
- onBind [-> DownloadService.java]
@Nullable
@Override
public IBinder onBind(Intent intent) {
log("bind Download Service");
startDispatch();
return mBinder;
}
複製代碼
- startDispatch() [-> DownloadService.java]
/** * start dispatch download queue. */
private void startDispatch() {
disposable = Observable
.create(new ObservableOnSubscribe<DownloadMission>() {
@Override
public void subscribe(ObservableEmitter<DownloadMission> emitter) throws Exception {
DownloadMission mission;
while (!emitter.isDisposed()) {
try {
log(WAITING_FOR_MISSION_COME);
mission = downloadQueue.take();
log(Constant.MISSION_COMING);
} catch (InterruptedException e) {
log("Interrupt blocking queue.");
continue;
}
emitter.onNext(mission);
}
emitter.onComplete();
}
})
.subscribeOn(Schedulers.newThread())
.subscribe(new Consumer<DownloadMission>() {
@Override
public void accept(DownloadMission mission) throws Exception {
mission.start(semaphore);
}
});
}
複製代碼
- 上述代碼就是下載任務分發線程的實現。其中
.subscribeOn(Schedulers.newThread())代表該線程經過 new Thread() 的方式產生的 - 下載任務用一個阻塞隊列來維護，阻塞隊列內部已經實現了同步機制，因此無需擔憂併發問題
- 只要沒有取消訂閱，該線程就會不停的嘗試從阻塞隊列中獲取下載任務併發射。若是隊列爲空，就會一直阻塞，直到有新任務入隊
- 上述代碼是不嚴謹的，對
disposable從新賦值前，沒有先嚐試對其取消訂閱。若是屢次調用bindService()，就會出現線程泄露
下載任務執行線程
顧名思義，就是下載任務的執行線程。該線程運行在 Schedulers.io() 線程池上。入參信號量用來限制同時下載的最大任務數。ui
- start(final Semaphore semaphore) [-> SingleMission.java]
@Override
public void start(final Semaphore semaphore) {
disposable = start(bean, semaphore, new MissionCallback() {
@Override
public void start() {
// 回調開始下載
if (callback != null) callback.start();
}
@Override
public void next(DownloadStatus value) {
// 回調下載中
status = value;
processor.onNext(started(value));
if (callback != null) callback.next(value);
}
@Override
public void error(Throwable throwable) {
// 回調下載失敗
processor.onNext(failed(status, throwable));
if (callback != null) callback.error(throwable);
}
@Override
public void complete() {
// 回調下載完成
processor.onNext(completed(status));
if (callback != null) callback.complete();
}
});
}
複製代碼
- start(DownloadBean bean, final Semaphore semaphore, final MissionCallback callback) [-> DownloadMission.java]
protected Disposable start(DownloadBean bean, final Semaphore semaphore, final MissionCallback callback) {
return rxdownload.download(bean)
.subscribeOn(Schedulers.io()) // 指定下載任務執行線程
.doOnLifecycle(new Consumer<Disposable>() {
@Override
public void accept(Disposable disposable) throws Exception {
if (canceled.get()) {
dispose(disposable);
}
log(TRY_TO_ACQUIRE_SEMAPHORE);
// 申請信號量
semaphore.acquire();
log(ACQUIRE_SUCCESS);
// 得到信號量後，需再次檢測是否已經暫停下載
if (canceled.get()) {
// 已經暫停，則取消訂閱，釋放信號量
dispose(disposable);
} else {
callback.start();
}
}
}, new Action() {
@Override
public void run() throws Exception {
// 取消訂閱時，須要釋放信號量
semaphore.release();
}
})
.subscribe(new Consumer<DownloadStatus>() {
@Override
public void accept(DownloadStatus value) throws Exception {
// 回調下載進度
callback.next(value);
}
}, new Consumer<Throwable>() {
@Override
public void accept(Throwable throwable) throws Exception {
// 回調下載失敗
callback.error(throwable);
}
}, new Action() {
@Override
public void run() throws Exception {
// 回調下載完成
callback.complete();
}
});
}
複製代碼
下載任務中斷線程
顧名思義，就是中斷下載任務的線程。包括暫停、刪除、所有暫停、所有取消四個操做。這些操做也運行在 Schedulers.io() 線程池上。this
- pauseServiceDownload(final String missionId) [-> RxDownlaod.java]
/** * Pause download. * <p> * Pause a url or all tasks belonging to missionId. * * @param missionId url or missionId */
public Observable<?> pauseServiceDownload(final String missionId) {
// createGeneralObservable 是一個異步綁定下載服務的Observable，經過資源數爲1的信號量實現強制同步
return createGeneralObservable(new GeneralObservableCallback() {
@Override
public void call() {
// 服務綁定後，調用服務的暫停下載
downloadService.pauseDownload(missionId);
}
}).observeOn(AndroidSchedulers.mainThread());
}
複製代碼
- createGeneralObservable(final GeneralObservableCallback callback) [-> RxDownload.java]
/** * return general observable * * @param callback Called when observable created. * @return Observable */
private Observable<?> createGeneralObservable(final GeneralObservableCallback callback) {
// 方法名起的很差，應該叫 bindService
return Observable.create(new ObservableOnSubscribe<Object>() {
@Override
public void subscribe(final ObservableEmitter<Object> emitter) throws Exception {
if (!bound) {
// 由於 onServiceConnected 是異步回調的，因此這裏用了個資源數爲1的信號量實現強制同步(CountDownLatch也能夠實現強制同步)
semaphore.acquire();
if (!bound) {
startBindServiceAndDo(new ServiceConnectedCallback() {
@Override
public void call() {
// 服務綁定後，回調 callback
doCall(callback, emitter);
// 釋放信號量
semaphore.release();
}
});
} else {
doCall(callback, emitter);
semaphore.release();
}
} else {
doCall(callback, emitter);
}
}
}).subscribeOn(Schedulers.io()); // 指定在 io 線程執行，因此暫停下載也是在這個線程執行
}
複製代碼
同理，刪除下載也會先調用 createGeneralObservable()，因此刪除操做也是在 Schedulers.io() 上執行的。url
總結
- 一個單獨的線程用來分發下載任務：
Schedulers.newThread() - 每次從線程池中取一個線程執行下載任務：
Schedulers.io() - 每次從線程池中取一個線程執行暫停、刪除下載任務：
Schedulers.io()
附
RxDownload2 系列文章：spa
相關文章
- 1. RxDownload2 源碼解析(三)
- 2. RxDownload2 源碼解析(一)
- 3. Picasso-源碼解析(二)
- 4. Vue源碼解析（二）
- 5. ReactNative iOS源碼解析（二）
- 6. LinkedList源碼解析（二）
- 7. V-LOAM源碼解析（二）
- 8. SpringMVC-源碼解析(二)
- 9. nacos 1.3.0源碼解析（二）
- 10. Spring源碼解析（二）
- 更多相關文章...
- • XML DOM 解析器 - XML DOM 教程
- • TCP報文格式解析 - TCP/IP教程
- • Java Agent入門實戰（二）-Instrumentation源碼概述
- • Scala 中文亂碼解決
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
歡迎關注本站公眾號,獲取更多信息
相關文章