作者:京东物流 王志明
在 App 开发中,经常会遇到处理异步任务的场景,如网络请求、读写文件等。Android、iOS 使用的是多线程,而在 Flutter 中为单线程事件循环,如下图所示
Dart 中有两个任务队列,分别为 microtask 队列和 event 队列,队列中的任务按照先进先出的顺序执行,而 microtask 队列的执行优先级高于 event 队列。在 main 方法执行完毕后,会启动事件循环,首先将 microtask 队列中的任务逐个执行完毕,再去执行 event 队列中的任务,每一个 event 队列中的任务在执行完成后,会再去优先执行 microtask 队列中的任务,如此反复,直到清空所有队列,这个过程就是 Dart 事件循环的处理机制。这种机制可以让我们更简单的处理异步任务,不用担心锁的问题。我们可以很容易的预测任务执行的顺序,但无法准确的预测到事件循环何时会处理到你期望执行的任务。例如创建了一个延时任务,但排在前面的任务结束前是不会处理这个延时任务的,也就说这个任务的等待时间可能会大于指定的延迟时间。
Dart 中的方法一旦开始执行就不会被打断,而 event 队列中的事件还来自于用户输入、IO、定时器、绘制等,这意味着在两个队列中都不适合执行计算量过大的任务,才能保证流畅的 UI 绘制和用户事件的快速响应。而且当一个任务的代码发生异常时,只会打断当前任务,后续任务不受影响,程序更不会退出。从上图还可以看出,将一个任务加入 microtask 队列,可以提高任务优先级,但是一般不建议这么做,除非比较紧急的任务并且计算量不大,因为 UI 绘制和处理用户事件是在 event 事件队列中的,滥用 microtask 队列可能会影响用户体验。
总结下 Dart 事件循环的主要概念:
可以使用顶层方法 scheduleMicrotask 或者 Future.microtask 方法,如下所示:
1 2 | scheduleMicrotask(() = > print ( 'microtask1' )); Future.microtask(() = > print ( 'microtask2' )); |
使用 Future.microtask 的优势在于可以在 then 回调中处理任务返回的结果。
1 | Future(() = > print ( 'event task' )); |
基于以上理论,通过如下代码可以验证 Dart 的事件循环机制:
1 2 3 4 5 6 7 8 9 10 11 12 | void main() { print ( 'main start' ); Future(() = > print ( 'event task1' )); Future.microtask(() = > print ( 'microtask1' )); Future(() = > print ( 'event task1' )); Future.microtask(() = > print ( 'microtask2' )); print ( 'main stop' ); |
执行结果:
1 2 3 4 5 6 | main start main stop microtask1 microtask2 event task1 event task1 |
通过输出结果可以看到,任务的执行顺序并不是按照编写代码的顺序来的,将任务添加到队列不会立刻执行,而执行顺序也完全符合前面讲的规则,当前 main 方法中的代码执行完毕后,才会去执行队列中的任务,且 microTask 队列的优先级高于 event 队列。
在 Dart 中通过 Future 来执行异步任务, Future 是对异步任务状态的封装,对任务结果的代理,通过 then 方法可以注册处理任务结果的回调方法。
创建方法 Future 方式:
Future()
Future.delayed()
Future.microtask()
Future.sync()
1 2 3 4 5 6 7 8 9 10 11 | factory Future(FutureOr<T> computation()) { _Future<T> result = new _Future<T>(); Timer.run(() { try { result._complete(computation()); } catch (e, s) { _completeWithErrorCallback(result, e, s); } }); return result; } |
上面是 Future() 的源码,可以看到内部是通过启动一个没有延迟的计时器来添加任务的,实用 try catch 来捕获任务代码中可能出现的异常,我们可以在 catchError 回调中来处理异常。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | factory Future.delayed(Duration duration, [FutureOr<T> computation()?]) { if (computation = = null && !typeAcceptsNull<T>()) { throw ArgumentError.value(null, "computation" , "The type parameter is not nullable" ); } _Future<T> result = new _Future<T>(); 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() 的区别是通过一个延迟的计时器来添加任务。
1 2 3 4 5 6 7 8 9 10 11 | factory Future.microtask(FutureOr<T> computation()) { _Future<T> result = new _Future<T>(); scheduleMicrotask(() { try { result._complete(computation()); } catch (e, s) { _completeWithErrorCallback(result, e, s); } }); return result; } |
Future.microtask() 是将任务添加到 microtask 队列,通过这种可以很方便通过 then 方法中的回调来处理任务的结果。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 | factory Future.sync(FutureOr<T> computation()) { try { var result = computation(); if (result is Future<T>) { return result; } else { / / TODO( 40014 ): Remove cast when type promotion works. return new _Future<T>.value(result as dynamic); } } catch (error, stackTrace) { var future = new _Future<T>(); 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() 中的任务会被立即执行,不会添加到任何队列。
在第一个章节中讲到了可以很容易的预测任务的执行顺序,下面我们通过一个例子来验证:
1 2 3 4 5 6 7 8 9 10 11 12 13 | 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' ); } |
执行结果:
1 2 3 4 5 6 7 | main start main stop microtask1 microtask2 event1 event2 delayed event |
因为代码比较简单,通过代码可以很容易的预测到执行结果,下面将复杂度稍微提高。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 | 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' ); } |
执行结果:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 | 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 |
看到结果后你可能会疑惑,为什么 event1、event1 - callback1、event1 - callback2 会连续输出,而 event2 - callback1 输出后为什么是 event3,event5、event5 - callback 为什么会在 microtask3 后输出?
这里我们补充下 then 方法的一些关键知识,理解了这些,上面的输出结果也就很好理解了:
1 2 3 4 5 6 7 8 9 | Future(() { throw 'error' ; }).then((_) { print ( 'success' ); }).catchError((error) { print (error); }).whenComplete(() { print ( 'completed' ); }); |
输出结果:
1 2 | error completed |
通过 catchError 方法注册的回调,可以用来处理任务代码产生的异常。不管 Future 中的任务执行成功与否,whenComplete 方法都会被调用。
使用 async、await 能以更简洁的编写异步代码,是 Dart 提供的一个语法糖。使用 async 关键字修饰的方法返回值类型为 Future,在 async 方法内可以使用 await 关键字来修饰异步任务,在方法内部达到同步执行的效果,可以达到简化代码和提高可读性的效果,不过如果想要处理异常,需要实用 try catch 语句来包裹 await 修饰的异步任务。
1 2 3 4 5 6 7 8 9 | void main() async { print (await getData()); } Future< int > getData() async { final a = await Future.delayed(Duration(seconds: 1 ), () = > 1 ); final b = await Future.delayed(Duration(seconds: 1 ), () = > 1 ); return a + b; } |
前面讲到耗时任务不适合放到 microtask 队列或 event 队列中执行,会导致 UI 卡顿。那么在 Flutter 中有没有既可以执行耗时任务又不影响 UI 绘制呢,其实是有的,前面提到 microtask 队列和 event 队列是在 main isolate 中运行的,而 isolate 是在线程中运行的,那我们开启一个新的 isolate 就可以了,相当于开启一个新的线程,使用多线程的方式来执行任务,Flutter 也为我们提供了相应的 Api。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | void main() async { compute<String, String>( getData, 'Alex' , ).then((result) { print (result); }); } String getData(String name) { / / 模拟耗时 3 秒 sleep(Duration(seconds: 3 )); return 'Hello $name' ; } |
compute 第一个参数是要执行的任务,第二个参数是要向任务发送的消息,需要注意的是第一个参数只支持顶层参数。使用 compute() 可以方便的执行耗时任务,但是滥用的话也会适得其反,因为每次调用,相当于新建一个 isolate。上面的代码执行一个经历了 isolate 的创建以及销毁过程,还有数据的传递会经历两次拷贝,因为 isolate 之间是完全隔离的,不能共享内存,整个过程除去任务本身的执行时间,也会非常的耗时,isolate 的创建也比较消耗内存,创建过多的 isolate 还有 OOM 的风险。这时我们就需要一个更优的解决方案,减少频繁创建销毁 isolate 所带来的消耗,最好是能创建一个类似于线程池的东西,只要提前初始化好,后面就可以随时使用,不用担心会发生前面所讲的问题,这时候 LoadBalancer 就派上用场了
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 | / / 用来创建 LoadBalancer Future<LoadBalancer> loadBalancerCreator = LoadBalancer.create( 2 , IsolateRunner.spawn); / / 全局可用的 loadBalancer late LoadBalancer loadBalancer; void main() async { / / 初始化 LoadBalancer loadBalancer = await loadBalancerCreator; / / 使用 LoadBalancer 执行任务 final result = await loadBalancer.run<String, String>(getData, 'Alex' ); print (result); } String getData(String name) { / / 模拟耗时 3 秒 sleep(Duration(seconds: 3 )); return 'Hello $name' ; } |
使用 LoadBalancer.create() 方法可以创建出一个 isolate 线程池,能够指定 isolate 的数量,并自动实现了负载均衡。应用启动后在合适的时机将其初始化好,后续就有一个全局可用的 LoadBalancer 了。
在开发中经常会有需要连续执行异步任务的场景,例如下面的例子,后面的一步任务直接需要以来前面任务的结果,所有任务正常执行完毕才算成功。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 | void main() async { print (await getData()); } Future< int > getData() { final completer = Completer< int >(); 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; } |
这种方式出现了回调地狱,代码非常难以阅读,实际开发中还会有处理异常的代码,会显得更加臃肿,编写难度也大,显然这种方式是不建议使用的。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | void main() async { print (await getData()); } Future< int > 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; }); } |
回调地狱的问题解决了,代码可读性提高很多。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | void main() async { print (await getData()); } Future< int > getData() async { int value = 0 ; value + = await Future(() = > 1 ); value + = await Future(() = > 2 ); value + = await Future(() = > 3 ); return value; } |
效果显而易见,代码更加清晰了。
在前面讲到了 Dart 方法执行时是不能被中断的,这就意味着一个 Future 任务开始后必然会走到完成的状态,但是很多时候我们需要又取消一个异步任务,唯一的办法就是在任务结束后不执行回调代码,就可以实现类似取消的效果。
在 Flutter 的 async 包中,提供了一个 CancelableOperation 给我们使用,使用它可以很简单的实现取消任务的需求。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | void main() async { / / 创建一个可以取消的任务 final cancelableOperation = CancelableOperation.fromFuture( Future(() async { print ( 'start' ); await Future.delayed(Duration(seconds: 3 )); / / 模拟耗时 3 秒 print ( 'end' ); }), onCancel: () = > print ( 'cancel...' ), ); / / 注册任务结束后的回调 cancelableOperation.value.then((val) { print ( 'finished' ); }); / / 模拟 1 秒后取消任务 Future.delayed(Duration(seconds: 1 )).then((_) = > cancelableOperation.cancel()); } |
CancelableOperation 是对 Future 的代理, 对 Future 的 then 进行了接管,判断 isCanceled 标记决定是否需要执行用户提供的回调。