深耕Flutter---扫清Flutter异步编程之路上的路障
Isolate
基本概念
Isolate更像是一个进程，全部的dart代码都会在上面运行，其内部有一个线程，用来处理event loop，还有一块内存，而且这块内存是私有的，也就说两个Isolate不共享内存（这也是和Java 的Thread的主要区别），要是多个Isolate想共同工做，只能经过port来发送消息进行通讯（通常会把特别耗时的操做，好比IO操做、图片压缩等容易引发main isolate卡顿的操做放到一个单独的Isolate中运行）。这样作的好处是分配内存和垃圾回收的时候不用对操做加锁，对于像flutter这种须要快速构建或者拆卸大量widget的应用来讲颇有帮助。
java
建立Isolate
两种方式建立：web
- 经过静态方法
Isolate.spawn建立（经常使用）
external static Future<Isolate> spawn<T>( void entryPoint(T message), T message, { bool paused: false, bool errorsAreFatal, SendPort onExit, SendPort onError, @Since("2.3") String debugName});
entryPoint入参必须是一个顶层函数或者一个static方法，不接受一个函数表达式或者一个实例的方法，不然会Invalid argument(s): Isolate.spawn expects to be passed a static or top-level function的错误。算法
- 经过构造函数建立（几乎用不到）
Isolate(this.controlPort, { this.pauseCapability, this.terminateCapability});
使用这种方法建立Isolate必须本身传入pause和terminate能力，不然调用pause和kill方法是没有任何效果的。编程
Isolate间通讯
- 不一样的Isolate之间通讯是经过
ReceivePort、SendPort来实现的。 - 能够在Isoalte之间传输的数据类型有基础类型(
null,num,bool,double,String)、SendPort、List、Map（List和Map元素的类型也必须是容许传输的类型，容许传递SendPort是实现双向通讯的基础），实践中发现也能传递一些属性为基础类型的简单对象，在1.17.1版本中甚至能传递future对象。 ReceivePort实现了Stream类，是一个非广播的Stream，因此只容许有一个监听者。- 经过看C的源码能够知道，每一个Isolate其实对应一个线程，其背后是有一个线程池的。
- 建立Isolate时会建立对应的堆空间、消息处理handler等等对应的信息。
单向通讯
void startOnWay() async { ReceivePort receivePort = ReceivePort(); Isolate isolate = await Isolate.spawn(oneWayEntryPoint, receivePort.sendPort); receivePort.listen((data) { print('$data'); receivePort.close();//若是不关的，它会一直监听 isolate?.kill(priority: Isolate.immediate);//杀死isolate isolate = null; }); } void oneWayEntryPoint(SendPort sendPort) { sendPort.send('hello world!'); }
双向通讯
void startBothWay() async { ReceivePort receivePort = ReceivePort(); await Isolate.spawn(entryPointBothWay, receivePort.sendPort); receivePort.listen((data) { if (data is SendPort) { data.send('hello child!'); } else { print('parent: $data'); } }); } void entryPointBothWay(SendPort sendPort) { ReceivePort r = ReceivePort(); sendPort.send(r.sendPort); r.listen((data) { print('child: $data'); sendPort.send('hello parent!'); }); }
compute函数
在flutter中实现isolate间通讯，能够直接使用更方便的API–compute函数。它其实是对isolate之间通讯的封装，每次调用后都会执行isolate.kill方法。demo以下：api
class Food { String name; double price; Food(this.name, this.price); } class AsyncDemo extends StatefulWidget { @override _AsyncState createState() => _AsyncState(); } class _AsyncState extends State<AsyncDemo> { var food = Food('apple', 23.1); @override Widget build(BuildContext context) { return Scaffold( body: Center( child: Text( ''' name: ${ food.name} price: ${ food.price} ''', style: TextStyle( fontSize: 22.0, ), ), ), floatingActionButton: FloatingActionButton(onPressed: () { compute<Food, Food>(handleCounter, food).then((value) { setState(() { food = value; }); }); }), ); } } Food handleCounter(Food food) { return food..price = 1.8; }
EventLoop
经过Isolate咱们知道dart app是个单线程app，那他是怎么实现异步操做的呢？答案是EventLoop和他相关联的两个队列–event queue、microtask queue。缓存
Event Queue
event queue包含几乎全部的事件，好比IO操做、绘制操做、多个isolate之间通讯、Future的部分操做等等。数据结构
Microtask Queue
这个队列用来执行用时很短的事件，这些事件会在把控制权交给Event Queue以前运行完毕。在实际开发中使用场景十分有限，整个Flutter源码里面总共有7处使用到了Microtask Queue（flutter中将事件提交到Microtask Queue上是使用 scheduleMicrotask api来操做的）。app
执行顺序
从上图能够看到，老是会先执行microtask，当microtask queue空的时候才会去执行event queue里的事件，每执行完一个event，都会从新去执行microtask queue里的事件。若是执行microtask里的事件消耗太多的时间，容易形成event queue阻塞，从而形成app的卡顿。
看下面demo，less
import 'dart:async'; main() { print('main #1 of 2'); scheduleMicrotask(() => print('microtask #1 of 3')); new Future.delayed(new Duration(seconds:1),// 1 () => print('future #1 (delayed)')); new Future(() => print('future #2 of 4')) // 2 .then((_) => print('future #2a')) .then((_) { print('future #2b'); scheduleMicrotask(() => print('microtask #0 (from future #2b)')); }) .then((_) => print('future #2c')); // X scheduleMicrotask(() => print('microtask #2 of 3')); new Future(() => print('future #3 of 4')) // 3 .then((_) => new Future( () => print('future #3a (a new future)')))// 5 .then((_) => print('future #3b'));// Y new Future(() => print('future #4 of 4')); // 4 scheduleMicrotask(() => print('microtask #3 of 3')); print('main #2 of 2'); }
输出：异步
main #1 of 2 main #2 of 2 // a microtask #1 of 3 microtask #2 of 3 microtask #3 of 3 // b future #2 of 4 future #2a future #2b // c future #2c microtask #0 (from future #2b) future #3 of 4 future #4 of 4 future #3a (a new future) future #3b future #1 (delayed)
分析：
程序会从上到下一次执行main方法里的同步方法，因此会依此输出到标 a的位置。scheduleMicrotask会把任务提交到microtask queue里，future会被提交到event queue里，因此先执行scheduleMicrotask的操做，对应输出中标记 b 的地方。microtaks queue里的事件都被处理完了，接下来，eventloop开始处理event queue里的事件。按道理来说代码里标记 1 2 3 4的future会被依此放到event queue队列中，可是标 1 的地方有一秒的延迟，因此标 1 的地方会在一秒后才会被放到event queue里，因此此时event queue里的事件排序是：2 3 4。先执行 2，依此输出到标记 c 的地方。这个时候看到了一个scheduleMicrotask，因此会把这个事件放到microtask的队列中去，此时这个future还未执行完，继续执行，输出future #2c。执行完毕后，event queue里的事件变成了 3 4，回去接着执行microtask queue里的事件，输出microtask #0 (from future #2b）。此时microtask queue里没有事件了，接着执行event queue里的事件，也就是事件 3，输出future #3 of 4，当执行到 5 的时候，又生成了一个新的future，放到了event queue里，事件3也就执行完了，此时event queue里的事件变成了 4(等待执行) 5(等待执行)。接着执行事件4，输出future #4 of 4，接着执行事件 5，依此输出future #3a (a new future) future #3b，最后通过一秒后事件1入队被执行，也就是最后输出的future #1 (delayed)。
须要注意的是代码中标记X和Y的地方，标记X的地方，因为他的上一步并无返回Future，因此标X的操做是在事件2后面进行的，而标记Y的地方，他上一步返回了一个Future对象，也就是事件 5，因此这个then函数里的操做是跟在事件5后面进行的。
Future
按官方文档上的介绍：它表明在未来某一时刻会获取到想要的值。当一个返回Future的函数被调用的时候，会发生两件事情：
- 这个Future对象会进去队列中排队等待被执行，此时会直接返回一个未完成状态的Future对象，可是这个函数是不进去队列的，它仅仅是被调用而已。
- 当值可得到的时候，Future处于一个叫 Completed with value 的状态，当执行失败时，处于一个叫作Completed with error的状态。
Future的三种状态：
要想使用Future，能够直接使用其提供的api，也可使用dart提供的语法糖async、await。
async、await
main() { fetchData(); } void fetchData() async { print('fetch data start'); await Future.delayed(Duration(seconds: 2)); print('Data is returned!'); } //输出： //fetch data start //Data is returned!(两秒后会输出)
须要注意的是，在dart 2.0 以前，async中的同步代码不会运行，碰见async就直接返回一个未完成的Future对象。当被async标记的函数没有返回值时，会返回一个Future< void>对象，有返回值的话，返回一个Future< T>对象。
Future API
Future()
void main() { var future = Future((){ return 'hello world!'; }); future.then(print); }
背后是经过Timer来实现的，直接看源码：
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.delayed
Future.delayed(Duration(seconds: 1), () { print('delayed 1'); });
1秒后输出相应值，其背后也是用的timer来实现，只不是传了一个duration对象过去。
factory Future.delayed(Duration duration, [FutureOr<T> computation()]) { _Future<T> result = new _Future<T>(); new Timer(duration, () { if (computation == null) { result._complete(null); } else { try { result._complete(computation()); } catch (e, s) { _completeWithErrorCallback(result, e, s); } } }); return result; }
Future.value
Future.value(1).then(print);
背后调用的源码以下（只考虑value非Future的状况下）：
void _asyncCompleteWithValue(T value) { _setPendingComplete(); _zone.scheduleMicrotask(() { _completeWithValue(value); }); }
能够看到背后并无调用Timer相关api，使用的是scheduleMicrotask，也就是说Future.value函数是在mictotask queue里完成的，能够经过下面的例子验证：
Future.delayed(Duration(seconds: 0)).then((value) => print('Future.delayed')); Future.value('Future.value').then(print); //输出： //Future.value ---优先执行microtask queue里的事件。 //Future.delayed
Future.sync
Future.sync(() => 'sync').then(print);
同步执行代码，下面是源码实现：
factory Future.sync(FutureOr<T> computation()) { try { var result = computation(); if (result is Future<T>) { return result; } else { 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; } }
要注意的是若是computation返回一个非future对象，会经过Future.value建立对象Future对象并返回，别忘了Future.value是在microtask 里完成的。再对比下下面两个例子的输出：
Demo1: Future.delayed(Duration(seconds: 0)).then((value) => print('delayed')); Future.sync(() => Future.delayed(Duration(seconds: 0))) .then((value) => print('sync')); //输出： //delayed //sync Demo2: Future.delayed(Duration(seconds: 0)).then((value) => print('delayed')); Future.sync(() => 'sync').then(print); //输出： //sync //delayed
Future.wait
final f1 = Future.delayed(Duration(seconds: 1)).then( (value) => 'first future.', ); final f2 = Future.delayed(Duration(seconds: 2)).then( (value) => 'second future.', ); Future.wait([f1, f2]).then((value) => print(value)); //输出： //[first future., second future.]
等待全部的future完成并返回一个包含全部future执行结果的list，一旦有一个future执行过程当中出现了error，最后的结果都不会输出。源码实现：
static Future<List<T>> wait<T>(Iterable<Future<T>> futures, { bool eagerError: false, void cleanUp(T successValue)}) { final _Future<List<T>> result = new _Future<List<T>>(); List<T> values; // Collects the values. Set to null on error. int remaining = 0; // How many futures are we waiting for. var error; // The first error from a future. StackTrace stackTrace; // The stackTrace that came with the error. // Handle an error from any of the futures. // TODO(jmesserly): use `void` return type once it can be inferred for the // `then` call below. handleError(Object theError, StackTrace theStackTrace) { remaining--; if (values != null) { if (cleanUp != null) { for (var value in values) { if (value != null) { // Ensure errors from cleanUp are uncaught. new Future.sync(() { cleanUp(value); }); } } } values = null; if (remaining == 0 || eagerError) { result._completeError(theError, theStackTrace); } else { error = theError; stackTrace = theStackTrace; } } else if (remaining == 0 && !eagerError) { result._completeError(error, stackTrace); } } try { // As each future completes, put its value into the corresponding // position in the list of values. for (var future in futures) { int pos = remaining; future.then((T value) { remaining--; if (values != null) { values[pos] = value; if (remaining == 0) { result._completeWithValue(values); } } else { if (cleanUp != null && value != null) { // Ensure errors from cleanUp are uncaught. new Future.sync(() { cleanUp(value); }); } if (remaining == 0 && !eagerError) { result._completeError(error, stackTrace); } } }, onError: handleError); // Increment the 'remaining' after the call to 'then'. // If that call throws, we don't expect any future callback from // the future, and we also don't increment remaining. remaining++; } if (remaining == 0) { return new Future.value(const []); } values = new List<T>(remaining); } catch (e, st) { // The error must have been thrown while iterating over the futures // list, or while installing a callback handler on the future. if (remaining == 0 || eagerError) { // Throw a new Future.error. // Don't just call `result._completeError` since that would propagate // the error too eagerly, not giving the callers time to install // error handlers. // Also, don't use `_asyncCompleteError` since that one doesn't give // zones the chance to intercept the error. return new Future.error(e, st); } else { // Don't allocate a list for values, thus indicating that there was an // error. // Set error to the caught exception. error = e; stackTrace = st; } } return result; }
总体来讲很简单，能够看到内部维护了一个future的列表，当所有执行完毕后返回结果，若是出错就直接返回错误信息。
Future.microtask
开启一个microtask的简单写法，用法同scheduleMicrotask，内部也是经过scheduleMicrotask来实现的，直接看源码吧：
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; }
异常处理
async、await的异常处理
async、await的异常处理和同步代码的异常处理是同样的，采用try catch的方法，一样也是可使用finally的：
main() { errorDemo(); } void errorDemo() async { try { await Future.delayed(Duration(seconds: 1)).then( (value) => throw '####Error####', ); } catch (e) { print(e); } finally { print('finally compeleted.'); } } // 输出： // ####Error#### // finally compeleted.
Future API 的异常处理
方法简介
Future api提供两种方式来处理异常，一个是Future<R> then<R>(FutureOr<R> onValue(T value), {Function onError}); 里的onError回调，另外一个是Future<T> catchError(Function onError, {bool test(Object error)});。
-
使用onError回调
main() { Future.delayed(Duration(seconds: 1), () { throw '#####Error#####'; }).then(print, onError: (error) { print('has error'); }).whenComplete(() => print('has complete!')); } // 输出： // has error // has complete!
能够看到并无报错，而是按照预期依此调用了onError、whenCompelte（至关于try catch里的 finally）回调。这有一个小细节须要注意的是，onError和onValue是同级的，当onValue里面发生异常时是不会走同级的onError回调的，onValue和onError是服务于一个future的。
-
使用catchError
Future.delayed(Duration(seconds: 1), () { throw '#####Error#####'; }).catchError((error){ print('has error'); }); // 输出： // has error
catchError和onError回调的用法差很少，最后也是输出了has error。看他的api，发现还有一个叫作test的可选命名参数，它是作什么的呢？只有当test返回true的时候才会走catchError回调，返回为false时会直接报错。他的入参是error对象，因此能够在test里面作一些特殊处理，好比error类型的过滤等等，默认状况下test始终返回true。
main() { Future.delayed(Duration(seconds: 1), () { throw '#####Error#####'; // 1 }).catchError((String error) { print('has error'); }, test: (error) { if (error is String) { return true; } return false; }); }
能够把标1的地方改为
throw 1再运行下试试效果。
onError和catchError的优先级问题
main() { Future.delayed(Duration(seconds: 1), () { // 1 throw '##### Error #####'; // 2 }).then(print, onError: (error) { // 3 print('onError callback'); // 4 }).catchError((error) { print('catchError callback'); }); } // 输出： // onError callback.
运行完delayed方法后，会返回一个Future，叫他Future1，标记 3 处的onError里的回调只对Future 1负责，Future 1爆了异常因此会走到onError回调里，也就是标记 4 的地方，一旦onError捕获了异常就不会再去执行catchError里的代码。若是说标 3 的地方并无设置onError回调，才会往catchError回调里走。
链式调用的异常处理
链式调用中，一旦某个环节爆了异常，下面的每个节点都会返回相同的错误，直到遇到catchError回调或者被下面的某个节点的onError回调拦截，直接看官网的一个例子吧：
Future<String> one() => new Future.value("from one"); Future<String> two() => new Future.error("error from two"); Future<String> three() => new Future.value("from three"); Future<String> four() => new Future.value("from four"); void main() { one() // Future completes with "from one". .then((_) => two()) // Future completes with two()'s error. .then((_) => three()) // Future completes with two()'s error. .then((_) => four()) // Future completes with two()'s error. .then((value) => Future.value(1)) // Future completes with two()'s error. .catchError((e) { print("Got error: ${e.error}"); // Finally, callback fires. return 42; // Future completes with 42. }).then((value) { print("The value is $value"); }); } // Output of this program: // Got error: error from two // The value is 42
同步、异步混合异常处理
不管是async、await，仍是Future，都是异步执行的，上面介绍的错误处理方式只能处理异步的异常，那么同步代码和异步代码混合的状况下该怎么处理异常呢？看下面的例子：
void main() { getLegalErpLength().then(print).catchError((error) { print('error happen!'); }); } Future<int> getLegalErpLength() { final erp = findErpFromDb(); // 1 return Future.value(erp).then(parseErp); } int parseErp(String erp) { return erp.length; } String findErpFromDb() { throw 'unknown error'; } //输出： Unhandled exception: unknown error #0 findErpFromDb #1 getLegalErpLength #2 main #3 _startIsolate.<anonymous closure> (dart:isolate-patch/isolate_patch.dart:301:19) #4 _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:168:12) Process finished with exit code 255
getLegalErpLength是一个异步方法，里面调用了同步方法：findErpFromDb，当findErpFromDb出现异常的时候getLegalErpLength时没法进行捕获的，因此直接输出了上面的错误栈。那若是想捕获findErpFromDb的异常应该怎么写呢？须要引入Future.sync ，修改以下：
Future<int> getLegalErpLength() { return Future.sync(() { final erp = findErpFromDb(); return Future.value(erp).then(parseErp); }); } // 输出： // error happen!
使用Future.sync将同步方法和异步方法包装一下，这样就完美的解决了上面的问题。
Future原理简介
-
Future()、Future.delayed()
背后使用的timer来实现，具体的源码能够参考dart sdk中的timer_impl.dart文件。简单来说，是用一个_TimerHeap的数据结构来存取timer，内部采用堆排序算法对其进行排序，当预先设置的时间到时，有一个eventhandler的东西负责处理设置的回调。
-
Future.value()、Future.microtask()
内部层层调用后，最终调用scheduleMicrotask方法，将其放到microtask queue中执行。内部也维护了一个loop，将每个回调封装成_AsyncCallbackEntry放到loop中，而后循环执行，直到loop为空。具体能够参考schedule_microtask.dart文件。
建立Future的另一种方式：Completer
completer能够用来建立future，相比使用future，completer能够本身指定future的完成时机。在开发中，能使用future的地方就直接使用future，它比completer更易读，除非有些状况下拿不到想要的future对象，可使用completer来替代，好比下面这个获取图片长宽的例子：
class CompleterDemo extends StatelessWidget { Widget build(BuildContext context) { Image image = new Image.network('https://img01.e23.cn/2020/0519/20200519022445869.jpg'); Completer<ui.Image> completer = new Completer<ui.Image>(); image.image.resolve(new ImageConfiguration()).addListener(ImageStreamListener( (image, synchronousCall) { completer.complete(image.image); }, )); return new Scaffold( appBar: new AppBar( title: new Text("Image Dimensions Example"), ), body: new ListView( children: [ new FutureBuilder<ui.Image>( future: completer.future, builder: (BuildContext context, AsyncSnapshot<ui.Image> snapshot) { if (snapshot.hasData) { return new Text( '${snapshot.data.width}x${snapshot.data.height}', style: Theme.of(context).textTheme.display3, ); } else { return new Text('Loading...'); } }, ), image, ], ), ); } }
Stream
相比future，stream能够用来处理数据流。Stream分为两种，一种是single subscription streams，只容许监听一次，另一种是Broadcast streams，能够容许监听屡次。直接看Stream的API你会发现，这东西和RxJava实在是太像了，一些API的名字和用法几乎差很少，因此这里再也不对Stream的经常使用API进行梳理，感兴趣的直接去看官方文档吧，对RxJava有了解的同窗学起来是很容易的。
建立Stream
-
Stream自带的一些factory构造函数，好比Stream.fromFuture(Future future)、Stream.fromIterable(Iterable elements)等等。以fromIterable为例：
Stream.fromIterable([1, 2]).listen((event) { print('event is $event'); }); // 输出： // event is 1 // event is 2
-
异步生成器函数：async* and yield（那对应的同步生成器函数使用的是 sync* and yield）
main() async { print('Start...'); await for (int number in generateNumbers(2)) { // 1 print(number); } print('End'); } Stream<int> generateNumbers(int maxCount) async* { print('yield output: '); for (int i = 0; i < maxCount; i++) { yield i; // 2 } print('yield* output: '); yield* createNumbers(maxCount); // 3 } Stream<int> createNumbers(int maxCount) async* { for (int i = 0; i <= maxCount; i++) { yield i; } } // 输出： // Start... // yield output: // 0 // 1 // yield* output: // 0 // 1 // 2 // End
须要注意两点：
一、异步生成器的产物Stream可使用await for进行遍历，await for必须实在async标记的函数中才会有效，另外同步生成器的产物Iterable可使用for进行遍历。
二、代码中标记2的地方使用的是 yield，标记3的地方使用的是yield*,他俩的区别是一个生成单个值，一个生成一串值，也就是一个Stream。
-
使用StreamController建立Stream
使用StreamController建立Stream能够在任何地方、任什么时候间添加event并处理event，可是和async相比要复杂的多，最重要的一点是async建立的Stream不会立刻执行，当第一个注册者注册的时候才会执行，可是利用StreamController建立的Stream则否则，看下面的代码：
main() async { var counterStream = timedCounter(const Duration(seconds: 1), 5); await Future.delayed(Duration(seconds: 3), () { print('3 seconds has gone.'); }); counterStream.listen((value) { print('$value -- ${DateTime.now()}'); }); } Stream<int> timedCounter(Duration interval, [int maxCount]) { var controller = StreamController<int>(); int counter = 0; void tick(Timer timer) { counter++; controller.add(counter); // Ask stream to send counter values as event. if (maxCount != null && counter >= maxCount) { timer.cancel(); controller.close(); // Ask stream to shut down and tell listeners. } } Timer.periodic(interval, tick); // BAD: Starts before it has subscribers. return controller.stream; } // 输出： 3 seconds has gone. 1 -- 2020-05-19 21:52:20.843943 2 -- 2020-05-19 21:52:20.847416 3 -- 2020-05-19 21:52:20.847438 4 -- 2020-05-19 21:52:21.831694 5 -- 2020-05-19 21:52:22.828021
能够看到前三个几乎是同时输出的，这是由于StreamController一旦建立Stream，里面的代码就开始执行，生成的数据会缓存下来，当有注册的时候立马把缓存输出给监听者，为了不这种状况，可使用StreamController各类回调方法进行限制，上面的代码修改以下：
Stream<int> timedCounter(Duration interval, [int maxCount]) { StreamController controller; int counter = 0; Timer timer; void tick(_) { counter++; controller.add(counter); if (maxCount != null && counter >= maxCount) { timer.cancel(); controller.close(); } } void startTimer() { timer = Timer.periodic(interval, tick); } controller = StreamController<int>( onListen: () { startTimer(); }, ); return controller.stream; } //输出： 3 seconds has gone. 1 -- 2020-05-19 22:02:32.488114 2 -- 2020-05-19 22:02:33.479319 3 -- 2020-05-19 22:02:34.477549 4 -- 2020-05-19 22:02:35.476372 5 -- 2020-05-19 22:02:36.481389
StreamController还有onCancel、onPause、onResume三个回调，用来处理取消、暂停、继续等操做。
Stream原理简介
Stream背后实现采用的是scheduleMicrotask，也就是说经过Stream发送的事件会直接被添加到microtask event queue里面。下面来跟踪下源码：
//StreamController里定义的add方法： void add(T event); //具体在_StreamController里面实现： void _add(T value) { if (hasListener) { _sendData(value); } else if (_isInitialState) { _ensurePendingEvents().add(new _DelayedData<T>(value)); } } //容许建立同步的和异步的StreamController，咱们直接来看异步的实现： abstract class _AsyncStreamControllerDispatch<T> implements _StreamController<T> { void _sendData(T data) { _subscription._addPending(new _DelayedData<T>(data)); } } //继续往里跟，会首先放到pending里面，而后判断是否为暂停状态，若是不是的话直接调用pending的schedule方法 void _addPending(_DelayedEvent event) { _StreamImplEvents<T> pending = _pending; if (_pending == null) { pending = _pending = new _StreamImplEvents<T>(); } pending.add(event); if (!_hasPending) { _state |= _STATE_HAS_PENDING; if (!_isPaused) { _pending.schedule(this); } } } //schedule的实现，看到了熟悉的scheduleMicrotask方法。 void schedule(_EventDispatch<T> dispatch) { if (isScheduled) return; assert(!isEmpty); if (_eventScheduled) { assert(_state == _STATE_CANCELED); _state = _STATE_SCHEDULED; return; } scheduleMicrotask(() { //先判断状态，若是已经取消了，则中止处理。 int oldState = _state; _state = _STATE_UNSCHEDULED; if (oldState == _STATE_CANCELED) return; handleNext(dispatch); }); _state = _STATE_SCHEDULED; }
总结
异步编程在平时开发中十分常见，了解原理的重要性没必要多说，因此直接点个赞吧，双击六六六！
参考资料： https://web.archive.org/web/20170704074724/https://webdev.dartlang.org/articles/performance/event-loop https://web.archive.org/web/20170702215338/https://www.dartlang.org/articles/libraries/futures-and-error-handling https://web.archive.org/web/20170702213650/https://www.dartlang.org/tutorials/language/futures https://dart.dev/codelabs/async-await https://medium.com/dartlang/dart-asynchronous-programming-futures-96937f831137 https://medium.com/dartlang/dart-asynchronous-programming-isolates-and-event-loops-bffc3e296a6a