來(lái)源| OSCHINA 社區(qū)

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

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

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


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

Dart 中有兩個(gè)隊(duì)列來(lái)執(zhí)行任務(wù)：microtask 隊(duì)列和 event 隊(duì)列。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

在 Dart 中通過(guò) Future 來(lái)執(zhí)行異步任務(wù)， Future 是對(duì)異步任務(wù)狀態(tài)的封裝，對(duì)任務(wù)結(jié)果的代理，通過(guò) then 方法可以注冊(cè)處理任務(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)部是通過(guò)啟動(dòng)一個(gè)沒(méi)有延遲的計(jì)時(shí)器來(lái)添加任務(wù)的，實(shí)用 try catch 來(lái)捕獲任務(wù)代碼中可能出現(xiàn)的異常，我們可以在 catchError 回調(diào)中來(lái)處理異常。

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ū)別是通過(guò)一個(gè)延遲的計(jì)時(shí)器來(lái)添加任務(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 隊(duì)列，通過(guò)這種可以很方便通過(guò) then 方法中的回調(diào)來(lái)處理任務(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ù)會(huì)被立即執(zhí)行，不會(huì)添加到任何隊(duì)列。 在第一個(gè)章節(jié)中講到了可以很容易的預(yù)測(cè)任務(wù)的執(zhí)行順序，下面我們通過(guò)一個(gè)例子來(lái)驗(yàn)證：

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àn)榇a比較簡(jiǎn)單，通過(guò)代碼可以很容易的預(yù)測(cè)到執(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é)果后你可能會(huì)疑惑，為什么 event1、event1 - callback1、event1 - callback2 會(huì)連續(xù)輸出，而 event2 - callback1 輸出后為什么是 event3，event5、event5 - callback 為什么會(huì)在 microtask3 后輸出？ 這里我們補(bǔ)充下 then 方法的一些關(guān)鍵知識(shí)，理解了這些，上面的輸出結(jié)果也就很好理解了：

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

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

如果 Future 中的任務(wù)在 then 方法調(diào)用之前已經(jīng)執(zhí)行完畢了，那么會(huì)有一個(gè)任務(wù)被加入到 microtask 隊(duì)列中。這個(gè)任務(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

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

2.6 async、await

使用 async、await 能以更簡(jiǎn)潔的編寫(xiě)異步代碼，是 Dart 提供的一個(gè)語(yǔ)法糖。使用 async 關(guān)鍵字修飾的方法返回值類型為 Future，在 async 方法內(nèi)可以使用 await 關(guān)鍵字來(lái)修飾異步任務(wù)，在方法內(nèi)部達(dá)到同步執(zhí)行的效果，可以達(dá)到簡(jiǎn)化代碼和提高可讀性的效果，不過(guò)如果想要處理異常，需要實(shí)用 try catch 語(yǔ)句來(lái)包裹 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 介紹

前面講到耗時(shí)任務(wù)不適合放到 microtask 隊(duì)列或 event 隊(duì)列中執(zhí)行，會(huì)導(dǎo)致 UI 卡頓。那么在 Flutter 中有沒(méi)有既可以執(zhí)行耗時(shí)任務(wù)又不影響 UI 繪制呢，其實(shí)是有的，前面提到 microtask 隊(duì)列和 event 隊(duì)列是在 main isolate 中運(yùn)行的，而 isolate 是在線程中運(yùn)行的，那我們開(kāi)啟一個(gè)新的 isolate 就可以了，相當(dāng)于開(kāi)啟一個(gè)新的線程，使用多線程的方式來(lái)執(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) {
  // 模擬耗時(shí)3秒
  sleep(Duration(seconds: 3));
  return 'Hello $name';
}

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

3.2 LoadBalancer

// 用來(lái)創(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) {
  // 模擬耗時(shí)3秒
  sleep(Duration(seconds: 3));
  return 'Hello $name';
}

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

4 實(shí)用經(jīng)驗(yàn)

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

在開(kāi)發(fā)中經(jīng)常會(huì)有需要連續(xù)執(zhí)行異步任務(wù)的場(chǎng)景，例如下面的例子，后面的一步任務(wù)直接需要以來(lái)前面任務(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)地獄，代碼非常難以閱讀，實(shí)際開(kāi)發(fā)中還會(huì)有處理異常的代碼，會(huì)顯得更加臃腫，編寫(xiě)難度也大，顯然這種方式是不建議使用的。

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)地獄的問(wèn)題解決了，代碼可讀性提高很多。

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;
}

效果顯而易見(jiàn)，代碼更加清晰了。

4.4 取消任務(wù)

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

4.5 CancelableOperation

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

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

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

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

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