「Android」Bolts-更簡單的完成線程調度和任務管理

本文來自尚妝Android團隊 路遠
發表於尚妝github博客，歡迎訂閱！

尤塞恩·聖利奧·博爾特 Usain St Leo Bolt，牙買加短跑運動員，男子100米、男子200米以及男子400米接力賽的世界紀錄保持人，同時是以上三項賽事的連續三屆奧運金牌得主。

使用 Bolts 能夠將一個完整的操做拆分紅多個子任務，這些子任務能夠自由的拆分、組合和替換，每一個任務做爲整個任務鏈的一環能夠運行在指定線程中，同時既能從上行任務中獲取任務結果，又能夠向下行任務發佈當前任務的結果，而沒必要考慮線程之間的交互。

Bolts-Android Bolts 在 Android 下的實現
Bolts-ObjC Bolts 在 OC 下的實現
Bolts-Swift Bolts 在 Swift 下的實現

前言

一個關於線程調度的簡單需求，在子線程從網絡下載圖片，並返回下載的圖片，在主線程使用該圖片更新到 UI，同時返回當前 UI 的狀態 json，在子線程將 json 數據保存到本地文件，完成後在主線程彈出提示，這中間涉及到了 4 次線程切換，同時後面的任務須要前面任務完成後的返回值做爲參數。

使用 Thread + Handler 實現，線程調度很不靈活，代碼可讀性差，不美觀，擴展性差，錯誤處理異常麻煩。

String url = "http://www.baidu.com";
Handler handler = new Handler(Looper.getMainLooper());
new Thread(() -> {
    // 下載
    Bitmap bitmap = downloadBitmap(url);
    handler.post(() -> {
        // 更新 UI
        String json = updateUI(bitmap);
        new Thread(() -> {
            // 向存儲寫入UI狀態
            saveUIState(json);
            // 保存成功後，提示
            handler.post(() -> toastMsg("save finish."));
        }).start();
    });
}).start();

使用 RxJava 實現，線程調度很是靈活，鏈式調用，代碼清晰，擴展性好，有統一的異常處理機制，不過 Rx 是一個很強大的庫，若是隻用來作線程調度的話，Rx 就顯得有點過重了。

Observable.just(URL)
        // 下載
        .map(this::downloadBitmap)
        .subscribeOn(Schedulers.newThread())
        // 更新UI
        .observeOn(AndroidSchedulers.mainThread())
        .map(this::updateUI)
        // 存儲 UI 狀態
        .observeOn(Schedulers.io())
        .map(this::saveUIState)
        // 顯示提示
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(rst -> toastMsg("save to " + rst),
                // handle error
                Throwable::printStackTrace);

使用 bolts 實現，線程調度靈活，鏈式調用，代碼清晰，具備良好的擴展性，具備統一的異常處理機制，雖然沒有 Rx 那麼豐富的操做符，可是勝在類庫很是很是小，只有 38 KB。

Task
        .forResult(URL)
        // 下載
        .onSuccess(task -> downloadBitmap(task.getResult()), Task.BACKGROUND_EXECUTOR)
        // 更新UI
        .onSuccess(task -> updateUI(task.getResult()), Task.UI_THREAD_EXECUTOR)
        // 存儲UI狀態
        .onSuccess(task -> saveUIState(task.getResult()), Task.BACKGROUND_EXECUTOR)
        // 提示
        .onSuccess(task -> toastMsg("save to " + task.getResult()), Task.UI_THREAD_EXECUT
        // handle error
        .continueWith(task -> {
            if (task.isFaulted()) {
                task.getError().printStackTrace();
                return false;
            }
            return true;
        });

線程調度器

共有 4 種類型執行線程，將任務分發到指定線程執行，分別是

1. backgroud - 後臺線程池，能夠併發執行任務。

2. scheduled - 單線程池，只有一個線程，主要用來執行 delay 操做。

3. immediate - 即時線程，若是線程調用棧小於 15，則在當前線程執行，不然代理給 background。

4. uiThread - 針對 Android 設計，使用 Handler 發送到主線程執行。

backgroud

主要用來在後臺併發執行多任務

public static final ExecutorService BACKGROUND_EXECUTOR = BoltsExecutors.background();

在 Android 平臺下根據 CPU 核數建立線程池，其餘狀況下，建立緩存線程池。

background = !isAndroidRuntime()
    ? java.util.concurrent.Executors.newCachedThreadPool()
    : AndroidExecutors.newCachedThreadPool();

scheduled

主要用於任務之間作 delay 操做，並不實際執行任務。

scheduled = Executors.newSingleThreadScheduledExecutor();

immediate

主要用來簡化那些不指定運行線程的方法，默認在當前線程去執行任務，使用 ThreadLocal 保存每一個線程調用棧的深度，若是深度不超過 15，則在當前線程執行，不然代理給 backgroud 執行。

private static final Executor IMMEDIATE_EXECUTOR = BoltsExecutors.immediate();

// 關鍵方法
@Override
public void execute(Runnable command) {
  int depth = incrementDepth();
  try {
    if (depth <= MAX_DEPTH) {
      command.run();
    } else {
      BoltsExecutors.background().execute(command)
    }
  } finally {
    decrementDepth();
  }
}

uiThread

爲 Android 專門設計，在主線程執行任務。

public static final Executor UI_THREAD_EXECUTOR = AndroidExecutors.uiThread();

private static class UIThreadExecutor implements Executor {
  @Override
  public void execute(Runnable command) {
    new Handler(Looper.getMainLooper()).post(command);
  }
}

核心類

Task，最核心的類，每一個子任務都是一個 Task，它們負責本身須要執行的任務。每一個 Task 具備 3 種狀態 Result、Error 和 Cancel，分別表明成功、異常和取消。

Continuation，是一個接口，它就像連接子任務每一環的鎖釦，把一個個獨立的任務連接在一塊兒。

經過 Task - Continuation - Task - Continuation ... 的形式組成完整的任務鏈，順序在各自線程執行。

建立 Task

根據 Task 的 3 種狀態，建立簡單的 Task，會複用已有的任務對象

public static <TResult> Task<TResult> forResult(TResult value)

public static <TResult> Task<TResult> forError(Exception error)

public static <TResult> Task<TResult> cancelled()

使用 delay 方法，延時執行並建立 Task

public static Task<Void> delay(long delay)

public static Task<Void> delay(long delay, CancellationToken cancellationToken)

使用 whenAny 方法，執行多個任務，當任意任務返回結果時，保存這個結果

public static <TResult> Task<Task<TResult>> whenAnyResult(Collection<? extends Task<TResult>> tasks)

public static Task<Task<?>> whenAny(Collection<? extends Task<?>> tasks)

使用 whenAll 方法，執行多個任務，當所有任務執行完後，返回結果

public static Task<Void> whenAll(Collection<? extends Task<?>> tasks) 

public static <TResult> Task<List<TResult>> whenAllResult(final Collection<? extends Task<TResult>> tasks)

使用 call 方法，執行一個任務，同時建立 Task

public static <TResult> Task<TResult> call(final Callable<TResult> callable, Executor executor,
      final CancellationToken ct)

連接子任務

使用 continueWith 方法，連接一個子任務，若是前行任務已經執行完成，則當即執行當前任務，不然加入隊列中，等待。

public <TContinuationResult> Task<TContinuationResult> continueWith(
      final Continuation<TResult, TContinuationResult> continuation, final Executor executor,
      final CancellationToken ct)

使用 continueWithTask 方法，在當前任務以後連接另外一個任務鏈，這種作法是爲了知足那種將部分任務組合在一塊兒分離出去，做爲公共任務的場景，他接受將另一個徹底獨立的任務鏈，追加在當前執行的任務後面。

public <TContinuationResult> Task<TContinuationResult> continueWithTask(
      final Continuation<TResult, Task<TContinuationResult>> continuation, final Executor executor,
      final CancellationToken ct)

使用 continueWhile 方法連接子任務，與 continueWith 區別在於，他有一個 predicate 表達式，只有當表達式成立時，纔會追加子任務，這樣作是在執行任務前能夠作一個攔截操做，也是爲了避免破環鏈式調用的總體風格。

public Task<Void> continueWhile(final Callable<Boolean> predicate,
      final Continuation<Void, Task<Void>> continuation, final Executor executor,
      final CancellationToken ct)

使用 onSuccess 和 onSuccessTask 連接單個任務個任務鏈，區別於 continueWith 在於，onSuccess 方法，前行任務若是失敗了，後行的任務也會直接失敗，不會再執行，可是 continueWith 的各個子任務之間沒有關聯，就算前行任務失敗，後行任務也會執行。

public <TContinuationResult> Task<TContinuationResult> onSuccess(
      final Continuation<TResult, TContinuationResult> continuation, Executor executor,
      final CancellationToken ct)

取消任務

Task 沒有 cancel 方法，而是使用了 CancellationToken 做爲標記，任務執行以前會檢查這個標記，若是標記爲退出，則會直接退出任務。

CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
CancellationToken token =   cancellationTokenSource.getToken();
Task.call((Callable<String>) () -> null,
        Task.BACKGROUND_EXECUTOR,
        token);
// 取消任務
cancellationTokenSource.cancel();

異常的處理

關於異常的處理，整個機制下來，每一個任務做爲一個獨立的單位，異常會被統一捕捉，所以沒必要針對任務中的方法進行單獨的處理。

若是使用了 continueWith 連接任務，那麼當前任務的的異常信息，將會保存在當前 Task 中在下行任務中進行處理，下行任務也能夠不處理這個異常，直接執行任務，那麼這個異常就到這裏中止了，不會再向下傳遞，也就是說，只有下行任務才知道當前任務的結果，不論是成功仍是異常。

固然了，若是任務之間有關聯，因爲上行任務的異常極大可能形成當前任務的異常，那麼當前任務異常的信息，又會向下傳遞，可是上行任務的異常就到這裏爲止了。

若是使用 onSuccess 之類的方法，若是上行任務異常了，那麼下行任務根本不會執行，而是直接將異常往下面傳遞，直到被處理掉。

任務的分離和組合

咱們能夠將一個完整的操做細分紅多個任務，每一個任務都遵循單一職責的原則而儘可能簡單，這樣能夠在任務之間再穿插新的任務，或者將部分任務分離出來組合到一塊兒等。

擴展性

咱們能夠在兩個細分的任務之間添加一個新的操做，而不影響上行和下行任務，如咱們給文章開頭的需求中更新 UI 以前，將 Bitmap 先保存到本地。

Task
        .forResult(URL)
        // 下載
        .onSuccess(task -> downloadBitmap(task.getResult()), Task.BACKGROUND_EXECUTOR)
        // 保存在本地
        .onSuccess(task -> saveBitmapToFile(task.getResult()),Task.BACKGROUND_EXECUTOR)
        // 更新UI
        .onSuccess(task -> updateUI(task.getResult()), Task.UI_THREAD_EXECUTOR)
        ...

複用性

對一些公共的操做，能夠單獨分離成新的任務，當須要作相似操做時，便可複用這部份功能，如能夠將下載圖片並更新 UI、保存狀態並彈出提示 兩塊功能分離出來，做爲公共的任務。

// 下載圖片->更新UI
public Continuation<String, Task<String>> downloadImageAndUpdateUI() {
    return task ->
            Task.call(() -> downloadBitmap(task.getResult()), Task.BACKGROUND_EXECUTOR)
                    .continueWith(taskWithBitmap -> updateUI(taskWithBitmap.getResult()), Task.UI_THREAD_EXECUTOR);
}

// 保存狀態->提示信息
public Continuation<String, Task<Boolean>> saveStateAndToast() {
    return task ->
            Task.call(() -> saveUIState(task.getResult()), Task.BACKGROUND_EXECUTOR)
                    .continueWith(taskWithPath -> toastMsg("save to " + taskWithPath.getResult()));
}

使用分離的任務

Task
        .forResult(URL)
        .continueWithTask(downloadImageAndUpdateUI())
        .continueWithTask(saveStateAndToast())
        ...

總結

在 Task 中有一個 continuations 是當前任務後面追加的任務列表，噹噹前任務成功、異常或者取消時，會去執行列表中的後續任務。

一般狀況下，咱們使用鏈式調用構建任務鏈，結果就是一條沒有分支的任務鏈。

添加任務時 ：每次添加一個 Continuation，就會生成一個 Task，加到上行任務的 continuations 列表中，等待執行，同時返回當前的 Task，以便後面的任務能夠連接到當前任務後面。

執行任務時 ：當前任務執行完以後，結果可能有 3 種，都會被保存到當前的 Task 中，而後檢查 continuations 列表中的後續任務，而當前的 Task 就會做爲參數，傳遞到後續連接的任務中，來讓後面的任務得知上行任務的結果。