Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

小菜前段时间简单研究了一下 Dart 单线程实现异步的操作，今天继续学习 Dart 的事件机制的任务调度；

任务调度

Dart 是单线程的，一个 Flutter 程序由一个或多个 isolate 组成，默认的执行方法均是在 main isolate 中；一个 isolate 中包含一个 Event Loop 和一个 Task Queue，而 Task Queue 包含 MicroTask Queue 微事件队列和 Event Queue 事件队列两种；

Dart 的事件机制是根据任务调度优先级来实现的；其中将任务添加到 MicroTask Queue 微事件队列的方式有 scheduleMicrotask() 和 Future.microtask() 两种；而将任务添加到 Event Queue 事件队列一般通过 Future 的相关构造方法实现；

MicroTask Queue 微事件队列的执行优先级高于 Event Queue 事件队列，而小菜简单理解，两种均类似于 Android 线程中的异步操作，只是 MicroTask Queue 微事件队列优先级相对更高；

Dart 的事件执行顺序如图所示；

启动 app 后优先执行 main() 方法中的同步方法；
查看 MicroTask Queue 是否为空，若不为空，优先循环执行 MicroTask Queue 中的 MicroTask，直到队列为空；
若 MicroTask Queue 队列为空则查看 Event Queue 事件队列，若不为空，则循环执行 Event Queue 中的 Event 事件，直到队列为空；
等两个队列的任务均执行完成后结束；

Tips: 当任务队列执行 MicroTask Queue 微事件队列时，Event Queue 事件队列被卡住，即应用无法绘制图形，处理鼠标点击事件，无法对 I/O 事件做出反应等；

案例尝试

每个 isolate 有各自的内存块和 Event Loop 是互相隔离的，小菜只尝试单个 isolate 中的微事件队列和事件队列的执行顺序；

小菜优先尝试最基本的 Future 构造函数和 MicroTask 的执行顺序；

_taskQueue01() {
  Future(() => print('TaskQueue -> Future A'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask A'));
  Future.delayed(Duration(seconds: 2), () => print('TaskQueue -> Future.delayed B'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask B'));
  Future(() => print('TaskQueue -> Future C'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask C'));
}

根据上述执行顺序，首先执行 main() 中同步的日志输出 Click start -> end，之后优先执行 MicroTask 微事件 scheduleMicrotask A -> C；再之后执行添加到 EventTask 事件队列的 Future() 构造函数 Future A -> C；最后执行因延迟 2s 加入事件队列的 Future B；

小菜在【案例一】的基础上尝试添加 Future.microtask() 构造函数；

_taskQueue02() {
  Future(() => print('TaskQueue -> Future A'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask A'));
  Future.delayed(Duration(seconds: 2), () => print('TaskQueue -> Future.delayed B'));
  Future.microtask(() => print('TaskQueue -> Future.microtask B'));
  Future(() => print('TaskQueue -> Future C'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask C'));
}

简单了解 Future.microtask() 函数源码，其实就是封装了一个 scheduleMicrotask() 微事件，所以微事件队列执行顺序为 scheduleMicrotask A -> B -> C；

小菜在【案例二】的基础上尝试添加 Future.sync() 构造函数；

_taskQueue03() {
  Future(() => print('TaskQueue -> Future A'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask A'));
  Future.sync(() => print('TaskQueue -> Future.sync D'));
  Future.delayed(Duration(seconds: 2), () => print('TaskQueue -> Future.delayed B'));
  Future.microtask(() => print('TaskQueue -> Future.microtask B'));
  Future(() => print('TaskQueue -> Future C'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask C'));
}

Future.sync() 为同步方法，会立即执行，因此是在 main() 点击同步日志 start 和 end 之间的；

小菜尝试添加 Future.then() 级联方法；

_taskQueue04() {
  Future(() => print('TaskQueue -> Future A'))
      .then((_) => print('TaskQueue -> Future A -> then()01'))
      .then((_) => print('TaskQueue -> Future A -> then()02'));
  Future.sync(() => print('TaskQueue -> Future.sync D'))
      .then((_) => print('TaskQueue -> Future.sync D -> then()01'))
      .then((_) => print('TaskQueue -> Future.sync D -> then()02'));
  Future.delayed(Duration(seconds: 2), () => print('TaskQueue -> Future.delayed B'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()01'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()02'));
  Future.microtask(() => print('TaskQueue -> Future.microtask B'))
      .then((_) => print('TaskQueue -> Future.microtask B -> then()01'))
      .then((_) => print('TaskQueue -> Future.microtask B -> then()02'));
  Future(() => print('TaskQueue -> Future C'))
      .then((_) => print('TaskQueue -> Future C -> then()01'))
      .then((_) => print('TaskQueue -> Future C -> then()02'));
}

Future.then() 级联函数是按照顺序执行的，执行完第一个 then() 函数后才会执行第二个 then()；且小菜理解为 then() 函数会在 Future() 执行之后立即执行，可以看作是存放在微事件队列中；因此整体的执行顺序是 sync()-> sync().then() -> microtask() -> microtask().then() -> Future() -> Future().then() -> delayed() -> delayed().then()；

小菜在【案例三 + 四】的基础上添加 Future.whenComplete() 方法；

_taskQueue05() {
  Future(() => print('TaskQueue -> Future A'))
      .then((_) => print('TaskQueue -> Future A -> then()01'))
      .then((_) => print('TaskQueue -> Future A -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future A -> whenComplete()'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask A'));
  Future.sync(() => print('TaskQueue -> Future.sync D'))
      .then((_) => print('TaskQueue -> Future.sync D -> then()01'))
      .then((_) => print('TaskQueue -> Future.sync D -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future.sync D -> whenComplete()'));
  Future.delayed(Duration(seconds: 2), () => print('TaskQueue -> Future.delayed B'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()01'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future.delayed B -> whenComplete()'));
  Future.microtask(() => print('TaskQueue -> Future.microtask B'))
      .then((_) => print('TaskQueue -> Future.microtask B -> then()01'))
      .then((_) => print('TaskQueue -> Future.microtask B -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future.microtask B -> whenComplete()'));
  Future(() => print('TaskQueue -> Future C'))
      .then((_) => print('TaskQueue -> Future C -> then()01'))
      .then((_) => print('TaskQueue -> Future C -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future C -> whenComplete()'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask C'));
}

Future.whenComplete() 是在 Future() 及 then() 函数执行完成之后立即执行；小菜也简单看作是存放在微事件队列中；依照小菜的测试案例来看，sync 执行同步任务完成后，会循环遍历微事件队列，其中 scheduleMicrotask A 是在 sync() 函数之前加入 MicroTask Queue 中，因此是先执行 scheduleMicrotask A 再执行 sync() 之后的 then() 和 whenComplete() 方法；

小菜尝试在 then() 函数中执行新的事件队列；

_taskQueue06() {
  Future(() => print('TaskQueue -> Future A'))
      .then((_) {
        print('TaskQueue -> Future A -> then()01');
        return Future.delayed(Duration(seconds: 1), () => print('TaskQueue -> Future.delayed D'));
      })
      .then((_) => print('TaskQueue -> Future A -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future A -> whenComplete()'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask A'));
  Future.delayed(Duration(seconds: 2), () => print('TaskQueue -> Future.delayed B'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()01'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future.delayed B -> whenComplete()'));
  Future(() => print('TaskQueue -> Future C'))
      .then((_) {
        print('TaskQueue -> Future C -> then()01');
        return scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask C'));
      })
      .then((_) => print('TaskQueue -> Future C -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future C -> whenComplete()'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask B'));
}

    a. 首先执行 MicroTask Queue 微事件队列中的 scheduleMicrotask A -> B；
    b. 之后以 FIFO 先进先出的顺序执行 EventTask Queue 事件队列中 Task；先执行 Future A，之后是 A 中 then()01，此时小菜设置了一个 1s 的延迟 Future.delayed D，因为 then() 的级联函数是需要等前面的 then() 函数执行完成后才能执行；因此优先执行 EventTask Queue 中的 Future C；
    c. 执行完 Future C 之后执行 then()01，此时小菜设置了一个 MicroTask，依照小菜的理解，then() / whenComplete() 均存放在 MicroTask Queue 中，因此新设置的 MicroTask 会放置在 MicroTask Queue 末尾，等 then()02 和 whenComplete() 再执行
    d. 此时 EventTask Queue 事件队列中已执行完毕，在 1s 后添加了新的 Future.delayed D 并执行；
    e. 待 Future.delayed D 执行结束或，Future A 的第二个级联 then() 函数和 whenComplete() 开始执行；
    f. 最后执行延迟最久的 2s 的 Future.delayed B 及其 then() / whenComplete() 函数；

小菜在测试过程中，与【案例六】代码几乎一致，只是在 Future.then() 中调用 Future() 时少了 return；

_taskQueue07() {
  Future(() => print('TaskQueue -> Future A'))
      .then((_) {
        print('TaskQueue -> Future A -> then()01');
        Future.delayed(Duration(seconds: 1), () => print('TaskQueue -> Future.delayed D'));
      })
      .then((_) => print('TaskQueue -> Future A -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future A -> whenComplete()'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask A'));
  Future.delayed(Duration(seconds: 2), () => print('TaskQueue -> Future.delayed B'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()01'))
      .then((_) => print('TaskQueue -> Future.delayed B -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future.delayed B -> whenComplete()'));
  Future(() => print('TaskQueue -> Future C'))
      .then((_) {
        print('TaskQueue -> Future C -> then()01');
        scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask C'));
      })
      .then((_) => print('TaskQueue -> Future C -> then()02'))
      .whenComplete(() => print('TaskQueue -> Future C -> whenComplete()'));
  scheduleMicrotask(() => print('TaskQueue -> scheduleMicrotask B'));
}

      小菜测试发现，虽然只是少了两个 return 但是执行顺序却变化很大；
    a. 首先执行微事件队列中的 scheduleMicrotask A -> B
    b. 之后执行 EventTask Queue 中的 Future A，执行第一个 print(then()01) 之后，小菜设置了 Future.delayed D，因为无需返回，所以将 Future.delayed D 事件队列末位后第一个 then() 函数已完成执行，此时可以执行第二个 then()02 和 whenComplete() 函数；
    c. 继续执行 EventTask Queue 中的 Future C，执行第一个 print(then()01) 之后，小菜设置了 scheduleMicrotask() C，无需 return，将 scheduleMicrotask() C 添加到微事件队列末位，此时第一个 then() 函数以完成执行，执行微事件队列中的第二个 then()02 和 whenComplete() 函数，再之后执行微事件队列中最后的 scheduleMicrotask() C；
    d. 之后执行事件队列末位的 1s 的 Future.delayed D，此时事件队列已为空；
    e. 最后执行 2s 添加到事件队列中的 Future.delayed B 及其 then() / whenComplete() 函数；

汇总小结

Dart 的任务调度模型是单线程轮询，不是基于时间调度的；我们可以知道各个事件的调度顺序，但无法明确得知各个事件调度的时间；例如：延迟 2s 并非一定是 2s 后执行，需要等之前的任务调度结束后才会执行；
注意级联函数中的新的任务调度是否需要返回结果后再继续；
实际中尽量在 EventTask 中执行耗时操作，减少在 MicroTask 中执行；若任务耗时时间较长可以尝试用 isolate 开启新的异步线程执行；

Task Queue 案例尝试

小菜对任务调度的理解还不够深入，如有错误请多多指导！

来源：阿策小和尚