作为系列文章的第十一篇，本篇将非常全面带你了解 Flutter 中最关键的设计之一，深入原理帮助你理解 Stream 全家桶，这也许是目前 Flutter 中最全面的 Stream 分析了。
文章汇总地址：
Flutter 完整实战实战系列文章专栏
Flutter 番外的世界系列文章专栏
一、Stream 由浅入深
Stream 在 Flutter 是属于非常关键的概念，在 Flutter 中，状态管理除了 InheritedWidget 之外，无论 rxdart，Bloc 模式，flutter_redux ，fish_redux 都离不开 Stream 的封装，而事实上 Stream 并不是 Flutter 中特有的，而是 Dart 中自带的逻辑。
通俗来说，Stream 就是事件流或者管道，事件流相信大家并不陌生，简单的说就是：基于事件流驱动设计代码，然后监听订阅事件，并针对事件变换处理响应。
而在 Flutter 中，整个 Stream 设计外部暴露的对象主要如下图，主要包含了 StreamController 、Sink 、Stream 、StreamSubscription 四个对象。
1、Stream 的简单使用
如下代码所示，Stream 的使用并不复杂，一般我们只需要：
- 创建
StreamController ， - 然后获取
StreamSink 用做事件入口， - 获取
Stream 对象用于监听， - 并且通过监听得到
StreamSubscription 管理事件订阅，最后在不需要时关闭即可，看起来是不是很简单？
class DataBloc {
///定义一个Controller
StreamController<List<String>> _dataController = StreamController<List<String>>();
///获取 StreamSink 做 add 入口
StreamSink<List<String>> get _dataSink => _dataController.sink;
///获取 Stream 用于监听
Stream<List<String>> get _dataStream => _dataController.stream;
///事件订阅对象
StreamSubscription _dataSubscription;
init() {
///监听事件
_dataSubscription = _dataStream.listen((value){
///do change
});
///改变事件
_dataSink.add(["first", "second", "three", "more"]);
}
close() {
///关闭
_dataSubscription.cancel();
_dataController.close();
}
}在设置好监听后，之后每次有事件变化时， listen 内的方法就会被调用，同时你还可以通过操作符对 Stream 进行变换处理。
如下代码所示，是不是一股 rx 风扑面而来？
_dataStream.where(test).map(convert).transform(streamTransformer).listen(onData);而在 Flutter 中， 最后结合 StreamBuilder , 就可以完成 基于事件流的异步状态控件 了！
StreamBuilder<List<String>>(
stream: dataStream,
initialData: ["none"],
///这里的 snapshot 是数据快照的意思
builder: (BuildContext context, AsyncSnapshot<List<String>> snapshot) {
///获取到数据，为所欲为的更新 UI
var data = snapshot.data;
return Container();
});那么问题来了，它们内部究竟是如果实现的呢？原理是什么？各自的作用是什么？都有哪些特性呢？后面我们将开始深入解析这个逻辑 。
2、Stream 四天王
从上面我们知道，在 Flutter 中使用 Stream 主要有四个对象，那么这四个对象是如何“勾搭”在一起的？他们各自又担任什么责职呢？
首先如下图，我们可以从进阶版的流程图上看出 整个 Stream 的内部工作流程。
Flutter中 Stream 、StreamController 、StreamSink 和 StreamSubscription 都是 abstract 对象，他们对外抽象出接口，而内部实现对象大部分都是 _ 开头的如 _SyncStreamController 、ControllerStream 等私有类，在这基础上整个流程概括起来就是：
有一个事件源叫 Stream，为了方便控制 Stream ，官方提供了使用 StreamController 作为管理；同时它对外提供了 StreamSink 对象作为事件输入口，可通过 sink 属性访问; 又提供 stream 属性提供 Stream 对象的监听和变换，最后得到的 StreamSubscription 可以管理事件的订阅。
所以我们可以总结出：
- StreamController ：如类名描述，用于整个
Stream 过程的控制，提供各类接口用于创建各种事件流。 - StreamSink：一般作为事件的入口，提供如
add ，addStream 等。 - Stream：事件源本身，一般可用于监听事件或者对事件进行转换，如
listen 、where 。 - StreamSubscription：事件订阅后的对象，表面上用于管理订阅过等各类操作，如
cacenl 、pause ，同时在内部也是事件的中转关键。
回到 Stream 的工作流程上，在上图中我们知道， 通过 StreamSink.add 添加一个事件时， 事件最后会回调到 listen 中的 onData 方法，这个过程是通过 zone.runUnaryGuarded 执行的，这里 zone.runUnaryGuarded 是什么作用后面再说，我们需要知道这个 onData 是怎么来的？
如上图，通过源码我们知道：
- 1、
Stream 在 listen 的时候传入了 onData 回调，这个回调会传入到 StreamSubscription 中，之后通过 zone.registerUnaryCallback 注册得到 _onData 对象( 不是前面的 onData 回调哦 )。 - 2、
StreamSink 在添加事件是，会执行到 StreamSubscription 中的 _sendData 方法，然后通过 _zone.runUnaryGuarded(_onData, data); 执行 1 中得到的 _onData 对象，触发 listen 时传入的回调方法。
可以看出整个流程都是和 StreamSubscription 相关的，现在我们已经知道从 事件入口到事件出口 的整个流程时怎么运作的，那么这个过程是**怎么异步执行的呢？其中频繁出现的 zone 是什么？
3、线程
首先我们需要知道，Stream 是怎么实现异步的？
这就需要说到 Dart 中的异步实现逻辑了，因为 Dart 是 单线程应用 ，和大多数单线程应用一样，Dart 是以 消息循环机制 来运行的，而这里面主要包含两个任务队列，一个是 microtask 内部队列，一个是 event 外部队列，而 microtask 的优先级又高于 event 。
默认的在 Dart 中，如 点击、滑动、IO、绘制事件 等事件都属于 event 外部队列，microtask 内部队列主要是由 Dart 内部产生，而 Stream 中的执行异步的模式就是 scheduleMicrotask 了。
因为 microtask 的优先级又高于 event ，所以如果 microtask 太多就可能会对触摸、绘制等外部事件造成阻塞卡顿哦。
如下图，就是 Stream 内部在执行异步操作过程执行流程：
4、Zone
那么 Zone 又是什么？它是哪里来的？
在上一篇章中说过，因为 Dart 中 Future 之类的异步操作是无法被当前代码 try/cacth 的，而在 Dart 中你可以给执行对象指定一个 Zone，类似提供一个沙箱环境 ，而在这个沙箱内，你就可以全部可以捕获、拦截或修改一些代码行为，比如所有未被处理的异常。
那么项目中默认的 Zone 是怎么来的？在 Flutter 中，Dart 中的 Zone 启动是在 _runMainZoned 方法 ，如下代码所示 _runMainZoned 的 @pragma("vm:entry-point") 注解表示该方式是给 Engine 调用的，到这里我们知道了 Zone 是怎么来的了。
///Dart 中
@pragma('vm:entry-point')
// ignore: unused_element
void _runMainZoned(Function startMainIsolateFunction, Function userMainFunction) {
startMainIsolateFunction((){
runZoned<Future<void>>(····);
}, null);
}
///C++ 中
if (tonic::LogIfError(tonic::DartInvokeField(
Dart_LookupLibrary(tonic::ToDart("dart:ui")), "_runMainZoned",
{start_main_isolate_function, user_entrypoint_function}))) {
FML_LOG(ERROR) << "Could not invoke the main entrypoint.";
return false;
}那么 zone.runUnaryGuarded 的作用是什么？相较于 scheduleMicrotask 的异步操作，官方的解释是：在此区域中使用参数执行给定操作并捕获同步错误。 类似的还有 runUnary 、 runBinaryGuarded 等，所以我们知道前面提到的 zone.runUnaryGuarded 就是 Flutter 在运行的这个 zone 里执行已经注册的 _onData，并捕获异常。
5、异步和同步
前面我们说了 Stream 的内部执行流程，那么同步和异步操作时又有什么区别？具体实现时怎么样的呢？
我们以默认 Stream 流程为例子， StreamController 的工厂创建可以通过 sync 指定同步还是异步，默认是异步模式的。 而无论异步还是同步，他们都是继承了 _StreamController 对象，区别还是在于 mixins 的是哪个 _EventDispatch 实现：
-
_AsyncStreamControllerDispatch -
_SyncStreamControllerDispatch
上面这两个 _EventDispatch 最大的不同就是在调用 sendData 提交事件时，是直接调用 StreamSubscription 的 _add 方法，还是调用 _addPending(new _DelayedData<T>(data)); 方法的区别。
如下图， 异步执行的逻辑就是上面说过的 scheduleMicrotask， 在 _StreamImplEvents 中 scheduleMicrotask 执行后，会调用 _DelayedData 的 perform ，最后通过 _sendData 触发 StreamSubscription 去回调数据 。
6、广播和非广播。
在 Stream 中又非为广播和非广播模式，如果是广播模式中，StreamControlle 的实现是由如下所示实现的，他们的基础关系如下图所示：
-
_SyncBroadcastStreamController -
_AsyncBroadcastStreamController
广播和非广播的区别在于调用 _createSubscription 时，内部对接口类 _StreamControllerLifecycle 的实现，同时它们的差异在于：
- 在
_StreamController 里判断了如果 Stream 是 _isInitialState 的，也就是订阅过的，就直接报错 "Stream has already been listened to." ，只有未订阅的才创建 StreamSubscription 。 - 在
_BroadcastStreamController 中，_isInitialState 的判断被去掉了，取而代之的是 isClosed 判断，并且在广播中, _sendData 是一个 forEach 执行：
_forEachListener((_BufferingStreamSubscription<T> subscription) {
subscription._add(data);
});7、Stream 变换
Stream 是支持变换处理的，针对 Stream 我们可以经过多次变化来得到我们需要的结果。那么这些变化是怎么实现的呢？
如下图所示，一般操作符变换的 Stream 实现类，都是继承了 _ForwardingStream , 在它的内部的_ForwardingStreamSubscription 里，会通过上一个 Pre A Stream 的 listen 添加 _handleData 回调，之后在回调里再次调用新的 Current B Stream 的 _handleData 。
所以事件变化的本质就是，变换都是对 Stream 的 listen 嵌套调用组成的。
同时 Stream 还有转换为 Future , 如 firstWhere 、 elementAt 、 reduce 等操作符方法，基本都是创建一个内部 _Future 实例，然后再 listen 的回调用调用 Future 方法返回。
二、StreamBuilder
如下代码所示, 在 Flutter 中通过 StreamBuilder 构建 Widget ，只需提供一个 Stream 实例即可，其中 AsyncSnapshot 对象为数据快照，通过 data 缓存了当前数据和状态，那 StreamBuilder 是如何与 Stream 关联起来的呢？
StreamBuilder<List<String>>(
stream: dataStream,
initialData: ["none"],
///这里的 snapshot 是数据快照的意思
builder: (BuildContext context, AsyncSnapshot<List<String>> snapshot) {
///获取到数据，为所欲为的更新 UI
var data = snapshot.data;
return Container();
});
如上图所示， StreamBuilder 的调用逻辑主要在 _StreamBuilderBaseState 中，_StreamBuilderBaseState 在 initState 、didUpdateWidget 中会调用 _subscribe 方法，从而调用 Stream 的 listen，然后通过 setState 更新UI，就是这么简单有木有？
我们常用的 setState 中其实是调用了 markNeedsBuild ，markNeedsBuild 内部标记 element 为 diry ，然后在下一帧 WidgetsBinding.drawFrame 才会被绘制，这可以看出 setState 并不是立即生效的哦。
三、rxdart
其实无论从订阅或者变换都可以看出， Dart 中的 Stream 已经自带了类似 rx 的效果，但是为了让 rx 的用户们更方便的使用，ReactiveX 就封装了 rxdart 来满足用户的熟悉感，如下图所示为它们的对应关系：
在 rxdart 中， Observable 是一个 Stream，而 Subject 继承了 Observable 也是一个 Stream，并且 Subject 实现了 StreamController 的接口，所以它也具有 Controller 的作用。
如下代码所示是 rxdart 的简单使用，可以看出它屏蔽了外界需要对 StreamSubscription 和 StreamSink 等的认知，更符合 rx 历史用户的理解。
final subject = PublishSubject<String>();
subject.stream.listen(observerA);
subject.add("AAAA1");
subject.add("AAAA2"));
subject.stream.listen(observeB);
subject.add("BBBB1");
subject.close();这里我们简单分析下，以上方代码为例，
-
PublishSubject 内部实际创建是创建了一个广播 StreamController<T>.broadcast 。 - 当我们调用
add 或者 addStream 时，最终会调用到的还是我们创建的 StreamController.add。 - 当我们调用
onListen 时，也是将回调设置到 StreamController 中。 -
rxdart 在做变换时，我们获取到的 Observable 就是 this，也就是 PublishSubject 自身这个 Stream ，而 Observable 一系列的变换，也是基于创建时传入的 stream 对象，比如：
@override
Observable<S> asyncMap<S>(FutureOr<S> convert(T value)) =>
Observable<S>(_stream.asyncMap(convert));所以我们可以看出来，rxdart 只是对 Stream 进行了概念变换，变成了我们熟悉的对象和操作符，而这也是为什么 rxdart 可以在 StreamBuilder 中直接使用的原因。
所以，到这里你对 Flutter 中 Stream 有全面的理解了没？
自此，第十一篇终于结束了！(///▽///)
资源推荐
- Github ：github.com/CarGuo/
- 开源 Flutter 完整项目：github.com/CarGuo/GSYG…
- 开源 Flutter 多案例学习型习项目: github.com/CarGuo/GSYF…
- 开源 Fluttre 实战电子书项目：github.com/CarGuo/GSYF…
完整开源项目推荐：
- GSYGithubApp Flutter
- GSYGithubApp React Native
- GSYGithubAppWeex
