Flutter 学习之旅(四十六) Rxdart 学习

本来这张想继续说一下flutter_bloc 这个框架,这里面会涉及到 Rxdart 一些东西,如果不说一下没有连贯性,看起代码来特别牵强,由于我也是现学现卖,所以我所描述的Rxdart 的 api 基本都是flutter_bloc 所用到的,在介绍的同时 ,同时会说一下部分的源码,让大家看起来更直接

这里我使用的是 rxdart: ^0.24.1 这个版本的

我从网上看了很多关于 rxdart 的文章,里面多少都有提到 Observable 的说明,但是我在使用的时候找不到这个类,分别引入了原来的0.18.1的版本,还有新的0.24.1 这个版本,发现在新的版本当中其实是没有Observable 这个类的,那么所有的功能又回归到Stream 了,
但是在Rxdart中Subject 这个类相当于controller 的功能, 既包含 Stream 这个事件源 ,也包含Subscription 这个管理对象,

我对于rxdart理解其实就是对Stream的转换,那么既然是转换,使用Stream 和Subject 他们有什么区别呢

我们先来看一下Stream 的简单应用

  List<int> array = [1, 2, 3, 4, 5, 6, 7];
  Stream<int> stream=Stream.fromIterable(array).asBroadcastStream().map((event) => event*event);
  stream.interval(Duration(seconds: 1)).listen((event) {
    printString('listen1:$event');
  });
  stream.listen((event) { printString('listen2:$event');});

打印结果

I/flutter (25410): tian.shm =listen2:1
I/flutter (25410): tian.shm =listen2:4
I/flutter (25410): tian.shm =listen2:9
I/flutter (25410): tian.shm =listen2:16
I/flutter (25410): tian.shm =listen2:25
I/flutter (25410): tian.shm =listen2:36
I/flutter (25410): tian.shm =listen2:49
I/flutter (25410): tian.shm =listen1:1
I/flutter (25410): tian.shm =listen1:4
I/flutter (25410): tian.shm =listen1:9
I/flutter (25410): tian.shm =listen1:16
I/flutter (25410): tian.shm =listen1:25
I/flutter (25410): tian.shm =listen1:36
I/flutter (25410): tian.shm =listen1:49

很简单的应用,创建一个Stream,并将它置位广播, 再对数据做一下简单的操作, 第一个listen 是每一秒打印一个日志,第二个listen 则是直接打印日志,
那么我们介绍这个方法有什么意义吗,
这里主要是为了说明Stream的一些静态方法,就从这个Stream.fromIterable 来说,他是同步还是异步? 既然可以监听,那么我可以在后续添加数据吗,还有是否可以多次监听,这个在上面已经体现了,既然是Stream ,那么只有BroadCastStream 可以被多次监听的,所以如果想要多次监听,所有在添加数据后必须使用asBroadcastStream()后才可以

那么剩下的就来先看一下Stream.fromIterable 这个方法他到底是同步还是异步的,

factory Stream.fromIterable(Iterable<T> elements) {
    return new _GeneratedStreamImpl<T>(
        () => new _IterablePendingEvents<T>(elements));
}

class _GeneratedStreamImpl<T> extends _StreamImpl<T> {
  final _EventGenerator<T> _pending;
  bool _isUsed = false;
  _GeneratedStreamImpl(this._pending);

  StreamSubscription<T> _createSubscription(void onData(T data)?,
      Function? onError, void onDone()?, bool cancelOnError) {
    if (_isUsed) throw new StateError("Stream has already been listened to.");
    _isUsed = true;
    return new _BufferingStreamSubscription<T>(
        onData, onError, onDone, cancelOnError)
      .._setPendingEvents(_pending());
  }
}

看到这个_BufferingStreamSubscription类大家应该很不陌生, 他维护了一下消息队列,并提供了一个相当于沙盒的运行环境, 对于是否是异步则看这个消息是怎么处理,那么只要看一下_IterablePendingEvents这个类是否维护了 Microtask ,下面我们来看一下源码,

 void handleNext(_EventDispatch<T> dispatch) {
    var iterator = _iterator;
    if (iterator == null) {
      throw new StateError("No events pending.");
    }
    bool movedNext = false;
    try {
      if (iterator.moveNext()) {
        movedNext = true;
        dispatch._sendData(iterator.current);
      } else {
        _iterator = null;
        dispatch._sendDone();
      }
    } catch (e, s) {
      if (!movedNext) {
      }
      dispatch._sendError(e, s);
    }
  }

我们看到_IterablePendingEvents这个类中只是维护了一下这个队列,并么有异步的过程,也就是说是否是异步需要只要大家看一下实现这个消息传递的event 就可以了
在学习Stream 的过程中,我们知道添加数据的过程中使用的是Sink调用controller 来添加数据,最后使用的是_BufferingStreamSubscription来管理这个队列的,而我们得到的只是Stream,并没有添加数据的功能,所以这就是直接使用Stream 相对于Subject 的缺陷,

PublishSubject

PublishSubject<int> publishSubject=PublishSubject<int>();
  publishSubject.add(1);
  publishSubject.add(2);
  publishSubject.add(3);
  publishSubject.interval(Duration(seconds: 1)).listen((value) {
    printString('listen1:$value');
  });
  publishSubject.add(4);
  publishSubject.listen((value) {
    printString('listen2:$value');
  });
  publishSubject.add(5);
  publishSubject.add(6);
  publishSubject.add(7);

结果

I/flutter (27171): tian.shm =listen2:5
I/flutter (27171): tian.shm =listen2:6
I/flutter (27171): tian.shm =listen2:7
I/flutter (27171): tian.shm =listen1:4
I/flutter (27171): tian.shm =listen1:5
I/flutter (27171): tian.shm =listen1:6
I/flutter (27171): tian.shm =listen1:7

从这里可以看出来先添加进去的数据如果没有监听那么这个消息就被消费掉了,至于为什么,先来说一下PublishSubject他的controller,

class PublishSubject<T> extends Subject<T> {
  PublishSubject._(StreamController<T> controller, Stream<T> stream)
      : super(controller, stream);
  factory PublishSubject(
      {void Function() onListen, void Function() onCancel, bool sync = false}) {
    final controller = StreamController<T>.broadcast(
      onListen: onListen,
      onCancel: onCancel,
      sync: sync,
    );
    return PublishSubject<T>._(
      controller,
      controller.stream,
    );
  }
}

使用的是_AsyncBroadcastStreamController,默认是异步的, 记得在说BroadcastStreamController 的时候说过,每次调用listen 的时候则生成一个subscription,消息转发的时候遍历subscription

  void _sendData(T data) {
    for (var subscription = _firstSubscription;
        subscription != null;
        subscription = subscription._next) {
      subscription._addPending(new _DelayedData<T>(data));
    }
  }

从这里可以看到,如果没有添加监听,这个消息就浪费了,

BehaviorSubject

同样使用上面的代码,将PublishSubject 变成BehaviorSubject ,

  BehaviorSubject<int> publishSubject=BehaviorSubject<int>();
  publishSubject.add(1);
  publishSubject.add(2);
  publishSubject.add(3);
  publishSubject.interval(Duration(seconds: 1)).listen((value) {
    printString('listen1:$value');
  });
  publishSubject.add(4);
  publishSubject.listen((value) {
    printString('listen2:$value');
  });
  publishSubject.add(5);
  publishSubject.add(6);
  publishSubject.add(7);

打印结果,

I/flutter (27171): tian.shm =listen2:4
I/flutter (27171): tian.shm =listen2:5
I/flutter (27171): tian.shm =listen2:6
I/flutter (27171): tian.shm =listen2:7
I/flutter (27171): tian.shm =listen1:3
I/flutter (27171): tian.shm =listen1:4
I/flutter (27171): tian.shm =listen1:5
I/flutter (27171): tian.shm =listen1:6
I/flutter (27171): tian.shm =listen1:7

相对于使用 PublishSubject ,打印结果都多打印了添加监听之前的一个数据,至于为什么,我们来看一下,

  factory BehaviorSubject({
    void Function() onListen,
    void Function() onCancel,
    bool sync = false,
  }) {
    // ignore: close_sinks
    final controller = StreamController<T>.broadcast(
      onListen: onListen,
      onCancel: onCancel,
      sync: sync,
    );

    final wrapper = _Wrapper<T>();

    return BehaviorSubject<T>._(
        controller,
        Rx.defer<T>(_deferStream(wrapper, controller, sync), reusable: true),
        wrapper);
  }

在创建BehaviorSubject的时候,同时初始化了一个_Wrapper这个对象,而这个对象是什么呢,

class _Wrapper<T> {
  T latestValue;
  Object latestError;
  StackTrace latestStackTrace;
  bool latestIsValue = false, latestIsError = false;
  _Wrapper();
  _Wrapper.seeded(this.latestValue) : latestIsValue = true;
  void setValue(T event) {
    latestIsValue = true;
    latestIsError = false;
    latestValue = event;
    latestError = null;
    latestStackTrace = null;
  }
  void setError(Object error, [StackTrace stackTrace]) {
    latestIsValue = false;
    latestIsError = true;
    latestValue = null;
    latestError = error;
    latestStackTrace = stackTrace;
  }
}

这个_Wrapper保存了一些操作的数据,那么什么时候对这个数据做操作呢,

class BehaviorSubject<T> extends Subject<T> implements ValueStream<T> {
  ...
  @override
  void onAdd(T event) => _wrapper.setValue(event);
  ...
}

每次添加数据的时候,保存添加的数据,也就是说,多打印的这个数据就是这个_Wrapper保存的数据,
既然数据保存了,而Stream 对于数据时不可以操作的,那么如何将保存的数据添加到原有监听呢,这里就归功于BehaviorSubject他的Stream的延迟创建的过程Rx.defer<T>(_deferStream(wrapper, controller, sync), reusable: true),每次添加监听后重新创建这个Stream,将这个缓存的数据重新添加入Stream 中,至于是怎么实现的,来看一下源码,

class DeferStream<T> extends Stream<T> {
  final Stream<T> Function() _factory;
  final bool _isReusable;

  @override
  bool get isBroadcast => _isReusable;

  /// Constructs a [Stream] lazily, at the moment of subscription, using
  /// the [streamFactory]
  DeferStream(Stream<T> Function() streamFactory, {bool reusable = false})
      : _isReusable = reusable,
        _factory = reusable
            ? streamFactory
            : (() {
                Stream<T> stream;
                return () => stream ??= streamFactory();
              }());
  @override
  StreamSubscription<T> listen(void Function(T event) onData,
          {Function onError, void Function() onDone, bool cancelOnError}) =>
      _factory().listen(onData,
          onError: onError, onDone: onDone, cancelOnError: cancelOnError);
}

入参就是是传入一个构造stream 的方法,每次监听的时候则调用该方法重新创建Stream ,这样就可以将默认_Wrapper缓存的数据添加到Stream,从而实现每次缓存一个数据

ReplaySubject

看过了BehaviorSubject,后再看ReplaySubject后就比较简单了,就是将_Wrapper 变成了一个队列,这个队列可以设置数据个数的上限,再添加监听后,如果超过上限则删除第一个后再将新的数据添加入这个队列中,

添加数据的方法,

  @override
  void onAdd(T event) {
    if (_queue.length == _maxSize) {
      _queue.removeFirst();
    }

    _queue.add(_Event(false, event: event));
  }

Stream重新创建的方法,根据队列重新创建Stream

Rx.defer<T>(
        () => queue.toList(growable: false).reversed.fold(controller.stream,
            (stream, event) {
          if (event.isError) {
            return stream.transform(StartWithErrorStreamTransformer(
                event.errorAndStackTrace.error,
                event.errorAndStackTrace.stackTrace,
                sync));
          } else {
            return stream
                .transform(StartWithStreamTransformer(event.event, sync: sync));
          }
        }),
        reusable: true,
      )

我们再来一个简单的例子

  ReplaySubject<int> publishSubject=ReplaySubject<int>();
  publishSubject.add(1);
  publishSubject.add(2);
  publishSubject.add(3);
  publishSubject.interval(Duration(seconds: 1)).listen((value) {
    printString('listen1:$value');
  });
  publishSubject.add(4);
  publishSubject.listen((value) {
    printString('listen2:$value');
  });
  publishSubject.add(5);
  publishSubject.add(6);
  publishSubject.add(7);

打印结果,

I/flutter (27171): tian.shm =listen2:1
I/flutter (27171): tian.shm =listen2:2
I/flutter (27171): tian.shm =listen2:3
I/flutter (27171): tian.shm =listen2:4
I/flutter (27171): tian.shm =listen2:5
Reloaded 4 of 966 libraries in 726ms.
I/flutter (27171): tian.shm =listen2:6
I/flutter (27171): tian.shm =listen2:7
I/flutter (27171): tian.shm =listen1:1
I/flutter (27171): tian.shm =listen1:2
I/flutter (27171): tian.shm =listen1:3
I/flutter (27171): tian.shm =listen1:4
I/flutter (27171): tian.shm =listen1:5
I/flutter (27171): tian.shm =listen1:6
I/flutter (27171): tian.shm =listen1:7

如果给ReplaySubject 设置一个最大值,效果大家肯定能想象的到,这里就不多做介绍了,

关于rxdart的操作符,如果像我这么讲起来,感觉再写个几十篇都不行,因为每一个操作符都有他的意义,就不多做介绍了

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

最后附上demo 地址

https://github.com/tsm19911014/tsm_flutter