小菜上周刚学习了 Stream 的部分方法，对 Stream 流的认知还很浅显，今天继续学习基础的 Stream 方法；
Stream Types
Single-Subscription
Single-Subscription 为单一订阅流，是最常见的 Stream 类型，这种方式只能被监听一次，若多次监听会导致 error；
var data = [1, 2, '3.toString()', true, true, false, true, 6];Stream stream = Stream.fromIterable(data).distinct();stream .listen((event) => print('Liste1 -> Stream.fromIterable -> $event')) .onDone(() => print('onDone1 -> Stream.fromIterable -> done 结束'));stream .listen((event2) => print('Liste2 -> Stream.fromIterable -> $event2')) .onDone(() => print('onDone2 -> Stream.fromIterable -> done 结束'));
复制代码
Broadcast
Broadcast 为广播订阅流，允许任意数量的接收者，这种方式可以在同一时间设置多个不同的监听器同时监听，同时你也可以在取消上一个订阅后再次对其发起监听；单一订阅流可以通过 asBroadcastStream 方式创建广播流；
var data = [1, 2, '3.toString()', true, true, false, true, 6];Stream stream = Stream.fromIterable(data).distinct();stream = stream.asBroadcastStream();stream .listen((event) => print('Liste1 -> Stream.fromIterable -> $event')) .onDone(() => print('onDone1 -> Stream.fromIterable -> done 结束'));stream .listen((event2) => print('Liste2 -> Stream.fromIterable -> $event2')) .onDone(() => print('onDone2 -> Stream.fromIterable -> done 结束'));
复制代码
Listener
小菜在之前的测试 Code 中都使用了 Listener，当 Stream 事件发出时可以通过 Listener 监听其状态；
StreamSubscription<T> listen(void onData(T event), {Function onError, void onDone(), bool cancelOnError});
复制代码
StreamController
StreamController 是 Stream 的控制器，允许发送数据、错误事件、结束事件给 Stream；可以用来检测 Stream 是否已暂停，是否有订阅者，以及变更时获取的回调；小菜分析源码，Stream 很多方法都是通过 StreamController 进行创建的；
StreamController 默认是对单一数据流进行控制；
factory StreamController( {void onListen(), void onPause(), void onResume(), onCancel(), bool sync: false}) { return sync ? new _SyncStreamController<T>(onListen, onPause, onResume, onCancel) : new _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);}
复制代码
通过 skin 向 StreamController 中添加数据；
通过 StreamController.stream 设置 Listener 监听；
若不设置 StreamController.close() 则不会监听到 onDone 方法，且 close() 之后不可继续添加事件或监听事件；
使用 StreamController 与单纯的 Listener 方式一样，单一信息流不可多次监听，需要转成广播流；
如果 Stream 直接监听 onDone 方法和通过 listener 中监听 onDone 方法，以 Stream 直接监听方法为准；
小菜尝试用 sink 方式向 StreamController 中添加数据，目前最后一个 'test' 是在你 listen 之后设置，依旧可以正常监听，但此时没有监听到 onDone 方法；
var data = [1, 2, '3.toString()', true, true, false, true, 6];var controller = StreamController();for (int i = 0; i < data.length; i++) { controller.sink.add(data[i]);}controller.stream.listen( (event) => print('Liste1 -> Stream.fromIterable -> $event'), onDone: () => print('onDone -> Stream.fromIterable -> done 结束'));controller.sink.add('test');
复制代码
小菜继续尝试在 'test' 之前调用 StreamController.close() 此时 StreamController 监听的是 close() 之前的数据并提示不能在 close 之后继续添加事件，且可以监听到 onDone 方法；
var data = [1, 2, '3.toString()', true, true, false, true, 6];var controller = StreamController();for (int i = 0; i < data.length; i++) { controller.sink.add(data[i]);}controller.stream.listen( (event) => print('Liste1 -> Stream.fromIterable -> $event'), onDone: () => print('onDone -> Stream.fromIterable -> done 结束'));controller.close();controller.sink.add('test');
复制代码
小菜尝试通过 Stream 监听 onDone 和 Listener 中监听 onDone 两种方式，以 Stream 直接监听 onDone 方式为准；
var data = [1, 2, '3.toString()', true, true, false, true, 6];var controller = StreamController();for (int i = 0; i < data.length; i++) { controller.sink.add(data[i]);}controller.stream .distinct() .listen((event) => print('Liste1 -> Stream.fromIterable -> $event'), onDone: () => print('onDone -> Stream.fromIterable -> done 结束')) .onDone(() => print('onDone1 -> Stream.fromIterable -> done 结束'));controller.close();
复制代码
小菜测试 StreamController.stream 获取的便是 Stream 可以设置滤重等方法；对于单一订阅流来说，依旧不能进行多个 Listener 监听，需要转换为广播流进行多处监听；
var data = [1, 2, '3.toString()', true, true, false, true, 6];var controller = StreamController();for (int i = 0; i < data.length; i++) { controller.sink.add(data[i]);}Stream stream = controller.stream.distinct();stream.listen((event) => print('Liste1 -> Stream.fromIterable -> $event')) .onDone(() => print('onDone1 -> Stream.fromIterable -> done 结束'));stream.listen((event) => print('Liste2 -> Stream.fromIterable -> $event')) .onDone(() => print('onDone2 -> Stream.fromIterable -> done 结束'));controller.close();
复制代码
Stream stream = controller.stream.distinct().asBroadcastStream();stream.listen((event) => print('Liste1 -> Stream.fromIterable -> $event')) .onDone(() => print('onDone1 -> Stream.fromIterable -> done 结束'));stream.listen((event) => print('Liste2 -> Stream.fromIterable -> $event')) .onDone(() => print('onDone2 -> Stream.fromIterable -> done 结束'));controller.close();
复制代码
小菜对 Stream 及 StreamController 的研究依旧很浅显，需要在实际应用中探索，也为 BLoc 的学习做基础；如有错误请多多指导！
来源：阿策小和尚
评论