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

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

任务调度

      Dart 是单线程的,一个 Flutter 程序由一个或多个 isolate 组成,默认的执行方法均是在 main isolate 中;一个 isolate 中包含一个 Event Loop 和一个 Task Queue,而 Task Queue 包含 MicroTask Queue 微事件队列和 Event Queue 事件队列两种;
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

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

      MicroTask Queue 微事件队列的执行优先级高于 Event Queue 事件队列,而小菜简单理解,两种均类似于 Android 线程中的异步操作,只是 MicroTask Queue 微事件队列优先级相对更高;
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

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

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

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

案例尝试

      每个 isolate 有各自的内存块和 Event Loop 是互相隔离的,小菜只尝试单个 isolate 中的微事件队列和事件队列的执行顺序;
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

  1. 小菜优先尝试最基本的 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
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

  1. 小菜在【案例一】的基础上尝试添加 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
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

  1. 小菜在【案例二】的基础上尝试添加 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() 点击同步日志 startend 之间的;
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

  1. 小菜尝试添加 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()
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

  1. 小菜在【案例三 + 四】的基础上添加 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() 方法;
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

  1. 小菜尝试在 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,之后是 Athen()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()02whenComplete() 再执行;
    d. 此时 EventTask Queue 事件队列中已执行完毕,在 1s 后添加了新的 Future.delayed D 并执行;
    e. 待 Future.delayed D 执行结束或,Future A 的第二个级联 then() 函数和 whenComplete() 开始执行;
    f. 最后执行延迟最久的 2sFuture.delayed B 及其 then() / whenComplete() 函数;
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

  1. 小菜在测试过程中,与【案例六】代码几乎一致,只是在 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()02whenComplete() 函数;
    c. 继续执行 EventTask Queue 中的 Future C,执行第一个 print(then()01) 之后,小菜设置了 scheduleMicrotask() C,无需 return,将 scheduleMicrotask() C 添加到微事件队列末位,此时第一个 then() 函数以完成执行,执行微事件队列中的第二个 then()02whenComplete() 函数,再之后执行微事件队列中最后的 scheduleMicrotask() C
    d. 之后执行事件队列末位的 1sFuture.delayed D,此时事件队列已为空;
    e. 最后执行 2s 添加到事件队列中的 Future.delayed B 及其 then() / whenComplete() 函数;
Flutter 95: 图解 Dart 单线程实现异步处理之 Task Queue

汇总小结

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

      Task Queue 案例尝试


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

来源: 阿策小和尚

上一篇:如何通过7个步骤构建机器学习模型


下一篇:哈工大 信息安全实验 XSS跨站脚本攻击原理与实践