Flutter完整开发实践详解十一：全面深入理解Stream

terry 2年前 (2023-09-23) 阅读数 123 #移动小程序

1.Stream由浅入深

Stream在Flutter中，状态管理是一个非常核心的概念。在 Flutter 中，国家行政除了 InheritedWidget 之外，无论 rxdart、BlocBloc

、fish_redux离不开Stream的包装，其实Stream并不是Flutter独有的，而是Dart自带的逻辑。通俗地说，Stream是一个事件流或管道。我想每个人都熟悉事件流。简单来说：基于事件流驱动设计代码，然后监听订阅事件，目标事件转换处理程序响应。在Flutter中，整个Stream设计对外暴露的对象主要如下所示，主要包括Stream Controller❀❀、、 Stream 、Stream订阅四物。 1。Stream 的简单使用如下代码所示，Stream的使用并不复杂。通常我们只需要：创建StreamController, 然后获取Stream❝事件❀获取Stream 监控对象，

并通过监控获得Stream订阅管理事件订阅，最后在不需要的时候关闭它。是不是看起来非常简单呢？

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中，最终结合Stream Builder，可以完成基于事件流的异步状态控制！

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的内部工作流程。

内部队列主要由Dart组成，它内部生成Stream中的异步执行模式为planMicrotask。

因为微任务的优先级高于事件。哦。

如下图，是Stream内部异步操作流程的执行过程：

`4，Zone`

那么Zone是什么呢？它从哪里来的？

上一章提到过，因为Dart中的Future等异步操作无法用当前代码try/cacth进行，但Dart中可以指定对象Sone，类似于提供一个沙箱环境，在这个沙箱中你可以捕获、捕获或修改一些代码行为，例如所有未处理的异常。

那么项目中的标准Zone来自哪里？在 Flutter 中，Zone

在 Dart 中的 _runMainZoned 方法 ❀_ 中启动，如以下❀代码所示， Zoned by @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等，所以我们知道前面提到的runed 就是 Flutter运行的区域运行注册的_onData 并捕获异常 。

5。异步和同步

我们讲了Stream的内部执行流程，那么同步和异步操作有什么区别呢？它在实践中是如何运作的？

我们以标准的Stream流程为例。 Stream控制器的工厂创建可以通过sync指定同步或异步。默认为异步模式。无论异步还是同步，它们都继承了_StreamController对象。区别在于其中mixins是_EventDispatch实现：

_AsyncFuturechController♿Syncspatch

上面两者最大的区别_EventDispatch 是当你调用 sendData 提交事件，是直接调用Stream订阅

方法的区别_add _addPending(new _DelayedData （数据））； 方法。 如下图，上面提到的异步执行scheduleMicrotask的逻辑，在_Confucis❓scheduleMic之后的rotask是执行后，将被称为_DelayedData  的执行，最终触发Stream订阅ens回调数据。 
6。广播和非广播。
Stream既没有广播模式也没有非广播模式。如果是广播模式，StreamControlle的实现如下，他们的基本关系如下图所示： 
_SyncBroadcast Stream控制器BStream控制器  与广播的区别和非广播的区别在于，当调用_createSubscription时，内部接口类_Stream Cont rollLifecycle是实现上的区别，而_Stream Controller决定了如果 _StreamController是_isInitialState

，即已经订阅了，只是直接报错❀cius❀"❀"us❝变成❀"，只创建取消订阅Stream订阅.i

，在广播中，_sendData 是一个 forEach 实现：

  _forEachListener((_BufferingStreamSubscription<T> subscription) {
      subscription._add(data);
    });
复制代码

7，Stream变换

Stream支持变换处理。

那么如何实现这些改变呢？

如下图所示，一般算子改造Stream继承了实现类_转发Stream，里面是Stream❀这里，Pre A Stream会通过添加_handleData回调听♻，然后回拨。新的C-us_handleData。

所以事件变化的本质是转换是由对Streams 的嵌套调用组成。

同时Stream也转换为未来，例如first，elementAt、reduce等运算符方法基本上创建一个内部 _future 实例，然后通过调用 Future 方法返回 List 回调。

2。

如下代码所示，通过Stream Builder在Flutter中构建一个Widget，只需要提供一个Stream❀where AsyncSnapshot 对象是一个数据快照，当前数据和状态通过data进行缓存。那么Stream Builder与Stream有什么关系呢？

StreamBuilder<List<String>>(
    stream: dataStream,
    initialData: ["none"],
    ///这里的 snapshot 是数据快照的意思
    builder: (BuildContext context, AsyncSnapshot<List<String>> snapshot) {
      ///获取到数据，为所欲为的更新 UI
      var data = snapshot.data;
      return Container();
    });

复制代码

如上图所示，Stream Builder的调用逻辑主要在_Stream BuilderBaseState⓻BaseState⓻B和⓻B、处于 initState , didUpdateWidget 会调用 _subscribe 方法，从而调用 ❀Stream❝ listen ，然后通过更新 UI 设置状态，难道就这么简单吗？

经常使用的SETSTATE实际调用markneedsbuild，和。将在下一帧中绘制。可以看到，setState并没有立即生效。

`3。 rxdart`

其实无论是订阅还是转换，可以看到 Dart 中的Stream 已经自带了类似rxrx的效果，以便更方便rx的用户，ReactiveX封装了rxdart以满足用户的熟悉感。下图展示了它们的对应关系：

in rxdart 在 Observable 中是rxdart 和和

`Subject`

 继承  Observable也是一个Stream，并且Subject实现了s的接口作为控制器。 以下代码是rxdart的简单使用。可以看出，它屏蔽了外界对Stream订阅和Stream水槽等的需求，更符合x的理解。 
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❙♿。 
当我们调用add或addStream时，我们最终会调用的是❀t-role❀d。 。 
当我们调用onListen时，回调也设置为StreamController。
rxdart 当您执行转换时， Observable 我们得到这个，即 PublishSubject❀self❿us  ，而  可观察到

许多转变也是基于创建时传入的stream对象，例如：

  @override
  Observable<S> asyncMap<S>(FutureOr<S> convert(T value)) =>
      Observable<S>(_stream.asyncMap(convert));
复制代码

所以我们可以看到rxdart❀❀只是forus 进行了概念上的转变，已经成为已知的对象和操作符，这就是为什么rxdart可以直接在Stream Builder中使用。

那么，你对Flutter中的Stream有了全面的了解了吗？ lianmao deXiaoguo
链接：https：//juejin.im/post/5cc2acf86fb9a0321f042041
来源：nuggets Property：用于商业重印的nuggetsProperty，请与作者联系。非商业转载请注明来源。