來源| OSCHINA 社區(qū)

作者 |京東云開發(fā)者-京東物流 王志明

1 Dart 中的事件循環(huán)模型

在 App 開發(fā)中，經(jīng)常會遇到處理異步任務(wù)的場景，如網(wǎng)絡(luò)請求、讀寫文件等。Android、iOS 使用的是多線程，而在 Flutter 中為單線程事件循環(huán)，如下圖所示


Dart 中有兩個任務(wù)隊列，分別為 microtask 隊列和 event 隊列，隊列中的任務(wù)按照先進先出的順序執(zhí)行，而 microtask 隊列的執(zhí)行優(yōu)先級高于 event 隊列。在 main 方法執(zhí)行完畢后，會啟動事件循環(huán)，首先將 microtask 隊列中的任務(wù)逐個執(zhí)行完畢，再去執(zhí)行 event 隊列中的任務(wù)，每一個 event 隊列中的任務(wù)在執(zhí)行完成后，會再去優(yōu)先執(zhí)行 microtask 隊列中的任務(wù)，如此反復(fù)，直到清空所有隊列，這個過程就是 Dart 事件循環(huán)的處理機制。這種機制可以讓我們更簡單的處理異步任務(wù)，不用擔(dān)心鎖的問題。我們可以很容易的預(yù)測任務(wù)執(zhí)行的順序，但無法準(zhǔn)確的預(yù)測到事件循環(huán)何時會處理到你期望執(zhí)行的任務(wù)。例如創(chuàng)建了一個延時任務(wù)，但排在前面的任務(wù)結(jié)束前是不會處理這個延時任務(wù)的，也就說這個任務(wù)的等待時間可能會大于指定的延遲時間。 Dart 中的方法一旦開始執(zhí)行就不會被打斷，而 event 隊列中的事件還來自于用戶輸入、IO、定時器、繪制等，這意味著在兩個隊列中都不適合執(zhí)行計算量過大的任務(wù)，才能保證流暢的 UI 繪制和用戶事件的快速響應(yīng)。而且當(dāng)一個任務(wù)的代碼發(fā)生異常時，只會打斷當(dāng)前任務(wù)，后續(xù)任務(wù)不受影響，程序更不會退出。從上圖還可以看出，將一個任務(wù)加入 microtask 隊列，可以提高任務(wù)優(yōu)先級，但是一般不建議這么做，除非比較緊急的任務(wù)并且計算量不大，因為 UI 繪制和處理用戶事件是在 event 事件隊列中的，濫用 microtask 隊列可能會影響用戶體驗。 總結(jié)下 Dart 事件循環(huán)的主要概念：

Dart 中有兩個隊列來執(zhí)行任務(wù)：microtask 隊列和 event 隊列。

事件循環(huán)在 main 方法執(zhí)行完畢后啟動， microtask 隊列中的任務(wù)會被優(yōu)先處理。

microtask 隊列只處理來自 Dart 內(nèi)部的任務(wù)，event 隊列中有來自 Dart 內(nèi)部的 Future、Timer、isolate message，還有來自系統(tǒng)的用戶輸入、IO、UI 繪制等外部事件任務(wù)。

Dart 中的方法執(zhí)行不會被打斷，因此兩個隊列中都不適合用來執(zhí)行計算量大的任務(wù)。

一個任務(wù)中未被處理的異常只會打斷當(dāng)前任務(wù)，后續(xù)任務(wù)不受影響，程序更不會退出。

1.1 向 microtask 隊列中添加任務(wù)

可以使用頂層方法 scheduleMicrotask 或者 Future.microtask 方法，如下所示：

scheduleMicrotask(() => print('microtask1'));
Future.microtask(() => print('microtask2'));

使用 Future.microtask 的優(yōu)勢在于可以在 then 回調(diào)中處理任務(wù)返回的結(jié)果。

1.2 向 event 隊列中添加任務(wù)

Future(() => print('event task'));

基于以上理論，通過如下代碼可以驗證 Dart 的事件循環(huán)機制：

void main() {
  print('main start');

  Future(() => print('event task1'));

  Future.microtask(() => print('microtask1'));

  Future(() => print('event task1'));

  Future.microtask(() => print('microtask2'));

  print('main stop');

執(zhí)行結(jié)果：

main start
main stop
microtask1
microtask2
event task1
event task1

通過輸出結(jié)果可以看到，任務(wù)的執(zhí)行順序并不是按照編寫代碼的順序來的，將任務(wù)添加到隊列不會立刻執(zhí)行，而執(zhí)行順序也完全符合前面講的規(guī)則，當(dāng)前 main 方法中的代碼執(zhí)行完畢后，才會去執(zhí)行隊列中的任務(wù)，且 microTask 隊列的優(yōu)先級高于 event 隊列。

2 Dart 中的異步實現(xiàn)

在 Dart 中通過 Future 來執(zhí)行異步任務(wù)， Future 是對異步任務(wù)狀態(tài)的封裝，對任務(wù)結(jié)果的代理，通過 then 方法可以注冊處理任務(wù)結(jié)果的回調(diào)方法。 創(chuàng)建方法 Future 方式：
Future()
Future.delayed()
Future.microtask()
Future.sync()

2.1 Future()

factory Future(FutureOr computation()) {
  _Future result = new _Future();
  Timer.run(() {
    try {
      result._complete(computation());
    } catch (e, s) {
      _completeWithErrorCallback(result, e, s);
    }
  });
  return result;
}

上面是 Future () 的源碼，可以看到內(nèi)部是通過啟動一個沒有延遲的計時器來添加任務(wù)的，實用 try catch 來捕獲任務(wù)代碼中可能出現(xiàn)的異常，我們可以在 catchError 回調(diào)中來處理異常。

2.2 Future.delayed()

factory Future.delayed(Duration duration, [FutureOr computation()?]) {
  if (computation == null && !typeAcceptsNull()) {
    throw ArgumentError.value(null, "computation", "The type parameter is not nullable");
  }
  _Future result = new _Future();
  new Timer(duration, () {
    if (computation == null) {
      result._complete(null as T);
    } else {
      try {
        result._complete(computation());
      } catch (e, s) {
        _completeWithErrorCallback(result, e, s);
      }
    }
  });
  return result;
}

Future.delayed () 與 Future () 的區(qū)別是通過一個延遲的計時器來添加任務(wù)。

2.3 Future.microtask()

factory Future.microtask(FutureOr computation()) {
  _Future result = new _Future();
  scheduleMicrotask(() {
    try {
      result._complete(computation());
    } catch (e, s) {
      _completeWithErrorCallback(result, e, s);
    }
  });
  return result;
}

Future.microtask () 是將任務(wù)添加到 microtask 隊列，通過這種可以很方便通過 then 方法中的回調(diào)來處理任務(wù)的結(jié)果。

2.4 Future.sync()

factory Future.sync(FutureOr computation()) {
  try {
    var result = computation();
    if (result is Future) {
      return result;
    } else {
      // TODO(40014): Remove cast when type promotion works.
      return new _Future.value(result as dynamic);
    }
  } catch (error, stackTrace) {
    var future = new _Future();
    AsyncError? replacement = Zone.current.errorCallback(error, stackTrace);
    if (replacement != null) {
      future._asyncCompleteError(replacement.error, replacement.stackTrace);
    } else {
      future._asyncCompleteError(error, stackTrace);
    }
    return future;
  }
}

Future.sync () 中的任務(wù)會被立即執(zhí)行，不會添加到任何隊列。 在第一個章節(jié)中講到了可以很容易的預(yù)測任務(wù)的執(zhí)行順序，下面我們通過一個例子來驗證：

void main() {
  print('main start');

  Future.microtask(() => print('microtask1'));

  Future.delayed(new Duration(seconds:1), () => print('delayed event'));
  Future(() => print('event1'));
  Future(() => print('event2'));

  Future.microtask(() => print('microtask2'));

  print('main stop');
}

執(zhí)行結(jié)果：

main start
main stop
microtask1
microtask2
event1
event2
delayed event

因為代碼比較簡單，通過代碼可以很容易的預(yù)測到執(zhí)行結(jié)果，下面將復(fù)雜度稍微提高。

void main() {
  print('main start');

  Future.microtask(() => print('microtask1'));

  Future.delayed(new Duration(seconds:1), () => print('delayed event'));

  Future(() => print('event1'))
    .then((_) => print('event1 - callback1'))
    .then((_) => print('event1 - callback2'));

  Future(() => print('event2')).then((_) {
    print('event2 - callback1');
    return Future(() => print('event4')).then((_) => print('event4 - callback'));
  }).then((_) {
    print('event2 - callback2');
    Future(() => print('event5')).then((_) => print('event5 - callback'));
  }).then((_) {
    print('event2 - callback3');
    Future.microtask(() => print('microtask3'));
  }).then((_) {
    print('event2 - callback4');
  });

  Future(() => print('event3'));

  Future.sync(() => print('sync task'));

  Future.microtask(() => print('microtask2')).then((_) => print('microtask2 - callbak'));

  print('main stop');
}

執(zhí)行結(jié)果：

main start
sync task
main stop

microtask1
microtask2
microtask2 - callbak

event1
event1 - callback1
event1 - callback2

event2
event2 - callback1
event3

event4
event4 - callback

event2 - callback2
event2 - callback3
event2 - callback4

microtask3
event5
event5 - callback

delayed event

看到結(jié)果后你可能會疑惑，為什么 event1、event1 - callback1、event1 - callback2 會連續(xù)輸出，而 event2 - callback1 輸出后為什么是 event3，event5、event5 - callback 為什么會在 microtask3 后輸出？ 這里我們補充下 then 方法的一些關(guān)鍵知識，理解了這些，上面的輸出結(jié)果也就很好理解了：

then 方法中的回調(diào)并不是按照它們注冊的順序來執(zhí)行。

Future 中的任務(wù)執(zhí)行完畢后會立刻執(zhí)行 then 方法中的回調(diào)，并且回調(diào)不會被添加到任何隊列中。

如果 Future 中的任務(wù)在 then 方法調(diào)用之前已經(jīng)執(zhí)行完畢了，那么會有一個任務(wù)被加入到 microtask 隊列中。這個任務(wù)執(zhí)行的就是被傳入 then 方法中的回調(diào)。

2.5 catchError、whenComplete

Future(() {
  throw 'error';
}).then((_) {
  print('success');
}).catchError((error) {
  print(error);
}).whenComplete(() {
  print('completed');
});

輸出結(jié)果：

error
completed

通過 catchError 方法注冊的回調(diào)，可以用來處理任務(wù)代碼產(chǎn)生的異常。不管 Future 中的任務(wù)執(zhí)行成功與否，whenComplete 方法都會被調(diào)用。

2.6 async、await

使用 async、await 能以更簡潔的編寫異步代碼，是 Dart 提供的一個語法糖。使用 async 關(guān)鍵字修飾的方法返回值類型為 Future，在 async 方法內(nèi)可以使用 await 關(guān)鍵字來修飾異步任務(wù)，在方法內(nèi)部達到同步執(zhí)行的效果，可以達到簡化代碼和提高可讀性的效果，不過如果想要處理異常，需要實用 try catch 語句來包裹 await 修飾的異步任務(wù)。

void main() async {
  print(await getData());
}

Future getData() async {
  final a = await Future.delayed(Duration(seconds: 1), () => 1);
  final b = await Future.delayed(Duration(seconds: 1), () => 1);
  return a + b;
}

3 Isolate 介紹

前面講到耗時任務(wù)不適合放到 microtask 隊列或 event 隊列中執(zhí)行，會導(dǎo)致 UI 卡頓。那么在 Flutter 中有沒有既可以執(zhí)行耗時任務(wù)又不影響 UI 繪制呢，其實是有的，前面提到 microtask 隊列和 event 隊列是在 main isolate 中運行的，而 isolate 是在線程中運行的，那我們開啟一個新的 isolate 就可以了，相當(dāng)于開啟一個新的線程，使用多線程的方式來執(zhí)行任務(wù)，F(xiàn)lutter 也為我們提供了相應(yīng)的 Api。

3.1 compute

void main() async {
  compute(
    getData,
    'Alex',
  ).then((result) {
    print(result);
  });
}

String getData(String name) {
  // 模擬耗時3秒
  sleep(Duration(seconds: 3));
  return 'Hello $name';
}

compute 第一個參數(shù)是要執(zhí)行的任務(wù)，第二個參數(shù)是要向任務(wù)發(fā)送的消息，需要注意的是第一個參數(shù)只支持頂層參數(shù)。使用 compute () 可以方便的執(zhí)行耗時任務(wù)，但是濫用的話也會適得其反，因為每次調(diào)用，相當(dāng)于新建一個 isolate。上面的代碼執(zhí)行一個經(jīng)歷了 isolate 的創(chuàng)建以及銷毀過程，還有數(shù)據(jù)的傳遞會經(jīng)歷兩次拷貝，因為 isolate 之間是完全隔離的，不能共享內(nèi)存，整個過程除去任務(wù)本身的執(zhí)行時間，也會非常的耗時，isolate 的創(chuàng)建也比較消耗內(nèi)存，創(chuàng)建過多的 isolate 還有 OOM 的風(fēng)險。這時我們就需要一個更優(yōu)的解決方案，減少頻繁創(chuàng)建銷毀 isolate 所帶來的消耗，最好是能創(chuàng)建一個類似于線程池的東西，只要提前初始化好，后面就可以隨時使用，不用擔(dān)心會發(fā)生前面所講的問題，這時候 LoadBalancer 就派上用場了

3.2 LoadBalancer

// 用來創(chuàng)建 LoadBalancer
Future loadBalancerCreator = LoadBalancer.create(2, IsolateRunner.spawn);

// 全局可用的 loadBalancer
late LoadBalancer loadBalancer;

void main() async {
  // 初始化 LoadBalancer
  loadBalancer = await loadBalancerCreator;

  // 使用 LoadBalancer 執(zhí)行任務(wù)
  final result = await loadBalancer.run(getData, 'Alex');
  print(result);
}

String getData(String name) {
  // 模擬耗時3秒
  sleep(Duration(seconds: 3));
  return 'Hello $name';
}

使用 LoadBalancer.create（） 方法可以創(chuàng)建出一個 isolate 線程池，能夠指定 isolate 的數(shù)量，并自動實現(xiàn)了負載均衡。應(yīng)用啟動后在合適的時機將其初始化好，后續(xù)就有一個全局可用的 LoadBalancer 了。

4 實用經(jīng)驗

4.1 指定任務(wù)的執(zhí)行順序

在開發(fā)中經(jīng)常會有需要連續(xù)執(zhí)行異步任務(wù)的場景，例如下面的例子，后面的一步任務(wù)直接需要以來前面任務(wù)的結(jié)果，所有任務(wù)正常執(zhí)行完畢才算成功。

void main() async {
  print(await getData());
}

Future getData() {
  final completer = Completer();
  int value = 0;

  Future(() {
    return 1;
  }).then((result1) {
    value += result1;
    return Future(() {
      return 2;
    }).then((result2) {
      value += result2;
      return Future(() {
        return 3;
      }).then((result3) {
        value += result3;
        completer.complete(value);
      });
    });
  });

  return completer.future;
}

這種方式出現(xiàn)了回調(diào)地獄，代碼非常難以閱讀，實際開發(fā)中還會有處理異常的代碼，會顯得更加臃腫，編寫難度也大，顯然這種方式是不建議使用的。

4.2 使用 then 的鏈?zhǔn)秸{(diào)用

void main() async {
  print(await getData());
}

Future getData() {
  int value = 0;
  return Future(() => 1).then((result1) {
    value += result1;
    return Future(() => 2);
  }).then((result2) {
    value += result2;
    return Future(() => 3);
  }).then((result3) {
    value += result3;
    return value;
  });
}

回調(diào)地獄的問題解決了，代碼可讀性提高很多。

4.3 使用 async、await

void main() async {
  print(await getData());
}


Future getData() async {
  int value = 0;

  value += await Future(() => 1);
  value += await Future(() => 2);
  value += await Future(() => 3);

  return value;
}

效果顯而易見，代碼更加清晰了。

4.4 取消任務(wù)

在前面講到了 Dart 方法執(zhí)行時是不能被中斷的，這就意味著一個 Future 任務(wù)開始后必然會走到完成的狀態(tài)，但是很多時候我們需要又取消一個異步任務(wù)，唯一的辦法就是在任務(wù)結(jié)束后不執(zhí)行回調(diào)代碼，就可以實現(xiàn)類似取消的效果。

4.5 CancelableOperation

在 Flutter 的 async 包中，提供了一個 CancelableOperation 給我們使用，使用它可以很簡單的實現(xiàn)取消任務(wù)的需求。

void main() async {
  // 創(chuàng)建一個可以取消的任務(wù)
  final cancelableOperation = CancelableOperation.fromFuture(
    Future(() async {
      print('start');
      await Future.delayed(Duration(seconds: 3)); // 模擬耗時3秒
      print('end');
    }),
    onCancel: () => print('cancel...'),
  );

  // 注冊任務(wù)結(jié)束后的回調(diào)
  cancelableOperation.value.then((val) {
    print('finished');
  });

  // 模擬1秒后取消任務(wù)
  Future.delayed(Duration(seconds: 1)).then((_) => cancelableOperation.cancel());
}

CancelableOperation 是對 Future 的代理， 對 Future 的 then 進行了接管，判斷 isCanceled 標(biāo)記決定是否需要執(zhí)行用戶提供的回調(diào)。