Flutter 学习之旅(四十二) Flutter 状态 Stream 学习(二)

BroadcastStream

在上一篇我们发现 listen 这个方法只能被添加一次, 即 _varData 这个参数只有一个,是不是只要把他变成PendingData形式的队列, 每次回调的时候遍历这个队列就可以通知所有callback 回调呢,
在普通的StreamController 里判断了如果 Stream 是 _isInitialState 的，也就是订阅过的，就直接报错 "Stream has already been listened to." ，只有未订阅的才创建 StreamSubscription 。

而在BroadcastStreamController 中去掉了这个方法变成了close 判断

  StreamSubscription<T> _subscribe(void onData(T data)?, Function? onError,
      void onDone()?, bool cancelOnError) {
    if (isClosed) {
      return new _DoneStreamSubscription<T>(onDone);
    }
    var subscription = new _BroadcastSubscription<T>(
        this, onData, onError, onDone, cancelOnError);
    _addListener(subscription);
    if (identical(_firstSubscription, _lastSubscription)) {
      // Only one listener, so it must be the first listener.
      _runGuarded(onListen);
    }
    return subscription;
  }
  

////_sendData  方法 遍历了每个subscription,并添加到任务队列
  void _sendData(T data) {
    if (_isEmpty) return;
    if (_hasOneListener) {
      _state |= _BroadcastStreamController._STATE_FIRING;
      _BroadcastSubscription<T> firstSubscription =
          _firstSubscription as dynamic;
      firstSubscription._add(data);
      _state &= ~_BroadcastStreamController._STATE_FIRING;
      if (_isEmpty) {
        _callOnCancel();
      }
      return;
    }
    _forEachListener((_BufferingStreamSubscription<T> subscription) {
      subscription._add(data);
    });
  }

说到这里自己其实是有些质疑源码的,为什么我设置了两个监听创建了两个subscription,你就执行了subscription._add(data);两次,不能只添加一次而依次调用回调吗,带着这个疑问我后来我又仔细想了一下,是不是入参是可变的,不同的环境会造就不同的data,那么是不是可以将方法传入StreamSink 中呢,但是这也解释不了为什么添加多次,后来我仔细想了一下其实他们的目的都是为了维护一个在Microtask 中的队列,如果向我所描述的做法还需要增加一个callback队列,并将BroadCastStreamController 的subscription 由源码中的队列改为一个单利,两相抵消暂时这么做在性能上是没有差别的.

在上面我们说StreamSink 可以添加耗时的方法,那么在StreamSink 发送普通的String int List 数据有什么意义吗,只不过是在异步中转了一下,其实普通的数据可以作为状态通知来用,但是感觉这不是他的精髓,他的精髓应该是在异步执行耗时任务,onlisten得到结果,但是实际情况是如果将一个StreamSink 方法add 后,在listen又得到这个方法,并没有什么意义,后来又从网上找到了这样一句话

因为 microtask 的优先级又高于 event ，所以如果 microtask 太多就可能会对触摸、绘制等外部事件造成阻塞卡顿哦。

也就是说如果在microtask 这个任务中执行耗时方法可能造成卡顿,到此研究耗时任务就放弃了,

Stream 转换

  ///添加监听的方法 ,调用_createSubscription
  StreamSubscription<T> listen(void onData(T value)?,
      {Function? onError, void onDone()?, bool? cancelOnError}) {
    return _createSubscription(onData, onError, onDone, cancelOnError ?? false);
  }
  

///创建  _ForwardingStreamSubscription
  StreamSubscription<T> _createSubscription(void onData(T data)?,
      Function? onError, void onDone()?, bool cancelOnError) {
    return new _ForwardingStreamSubscription<S, T>(
        this, onData, onError, onDone, cancelOnError);
  }

  /// 转换的过程中将自身的_handleData 加入回调队列,
  _ForwardingStreamSubscription(this._stream, void onData(T data)?,
      Function? onError, void onDone()?, bool cancelOnError)
      : super(onData, onError, onDone, cancelOnError) {
    _subscription = _stream._source
        .listen(_handleData, onError: _handleError, onDone: _handleDone);
  }

///Stream 转换  通常是继承自 _ForwardingStream 来实现的,
class _WhereStream<T> extends _ForwardingStream<T, T> {
  final bool Function(T) _test;

  _WhereStream(Stream<T> source, bool test(T value))
      : _test = test,
        super(source);

///回调后,又继续添加了,这里看出来的嵌套
  void _handleData(T inputEvent, _EventSink<T> sink) {
    bool satisfies;
    try {
      satisfies = _test(inputEvent);
    } catch (e, s) {
      _addErrorWithReplacement(sink, e, s);
      return;
    }
    if (satisfies) {
      sink._add(inputEvent);
    }
  }
}

他的做法是添加listen ,在listen 回调的时候调用自身的StreamSink.add(),将自身的_handleData()添加到事件源,在执行回调后重新添加一次,所谓的转换只不过是嵌套了一层

所以事件变化的本质就是，变换都是对 Stream 的 listen 嵌套调用组成的。

昨天看了一些面试题,上面经常会问什么是Stream ,那么你心里的Stream 是什么呢,我来说一下我写完这两篇后我对Stream 的理解,

Stream 是流或者管道,提供了一种同步或异步以队列的方式处理微任务的消息的机制.

我学习flutter的整个过程都记录在里面了
https://www.jianshu.com/c/36554cb4c804

最后附上demo 地址

https://github.com/tsm19911014/tsm_flutter