Flutter Stream简介及部分操作符使用

前言

    对于刚接触Flutter的同学来说,Stream(流)是一个相对比较抽象，也相对比较难以理解的东西。准确的来说Stream并不是Flutter的特性，而是Dart语言自身所带库。Stream和Future都位于dart:async核心库，是Dart中异步操作的两大高手。所以不仅仅可以用于Flutter,而是可以用于任何Dart语言上的实现。

    在我们刚开始学习Flutter的时候基本都是使用 StatefulWidget和setState((){})来刷新界面的数据，当我熟练使用流之后就可以基本完全使用StatelessWidget告别 StatefulWidget同样达到数据刷新效果。

Stream分类

流可以分为两类：

• 单订阅流(Single Subscription),这种流最多只能有一个监听器(listener)
• 多订阅流(Broadcast),这种流可以有多个监听器监听（listener）

Stream创建方式

• Stream<T>.fromFuture 接收一个Future对象作为参数

  Stream<String>.fromFuture(getData());
  
  Future<String> getData() async{
      await Future.delayed(Duration(seconds: 5));
      return "返回一个Future对象";
    }

• Stream<T>.fromIterable 接收一个集合对象作为参数

  Stream<String>.fromIterable(['A','B','C']);

• Stream<T>.fromFutures 接收一个Future集合对象作为参数

  Stream<String>.fromFutures([getData()]);

• Stream<T>.periodic 接收一个 Duration对象作为参数

  Duration interval = Duration(seconds: 1);
  Stream<int> stream = Stream<int>.periodic(interval);

操作方式及使用

使用用 Stream<T> .periodic 方式创建一个Stream对象

用于创建一个周期发送事件的流，如下

///用于创建一个每隔一秒发送一次无限事件的流,并打印出值
void _stream() async{
    Duration interval = Duration(seconds: 1);
    Stream<int> stream = Stream.periodic(interval, (data) => data);
    await for(int i in stream ){
      print(i); 
    }
}

//(data){return data;} 上面的这句是用来回来值，如果省略这句，打印的值都为null

.

• Stream<T>.take(int count)

  上面创建了一个无限每隔一秒发送一次事件的流，如果我们想指定只发送10个事件则，用take。下面就只会打印出0-9

  void _stream() async{
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
      stream = stream.take(10); //指定发送事件个数
      await for(int i in stream ){
       print(i);
      }
  }

• Stream<T>.takeWhile

  上面这种方式我们是只制定了发送事件的个数，如果我们也不知道发送多少个事件，我们可以从返回的结果上做一个返回值的限制,上面结果也可以用以下方式实现

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
  //    stream = stream.take(10);
      stream = stream.takeWhile((data) {
        return data < 10;
      });
      await for (int i in stream) {
        print(i);
      }
    }

• Stream<T>.skip(int count)

  skip可以指定跳过前面的几个事件,如下会跳过0和1，输出 2-9;

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
      stream = stream.take(10);
      stream = stream.skip(2);
      await for (int i in stream) {
        print(i);
      }
    }

• Stream<T>.skipWhile

  可以指定跳过不发送事件的指定条件，如下跳过0-4的输出，输出5-9

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
      stream = stream.take(10);
      stream = stream.skipWhile((data) => data<5);
      await for (int i in stream) {
        print(i);
      }
    }

• Stream<T>.toList()

  将流中所有的数据收集存放在List中，并返回 Future<List<T>>对象，listData里面 0-9

  1.这个是一个异步方法，要结果则需要使用await关键字

  2.这个是等待Stream当流结束时，一次返回结果

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
      stream = stream.take(10);
      List<int> listData = await stream.toList();
      for (int i in listData) {
        print(i);
      }
    }

• Stream<T>. listen()

  这是一种特定的可以用于监听数据流的方式，和 forEach循环的效果一致，但是返回的是StreamSubscription<T>对象，如下也会输出0-9,同时打印出 ”流已完成“

  看一下源码这种方式可以接收

  StreamSubscription<T> listen(void onData(T event),
        {Function onError, void onDone(), bool cancelOnError});

  1.onData是接收到数据的处理,必须要实现的方法

  2.onError流发生错误时候的处理

  3.onDone流完成时候调取

  4.cancelOnError发生错误的时候是否立马终止

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
      stream = stream.take(10);
      stream.listen((data) {
        print(data);
      }, onError: (error) {
        print("流发生错误");
      }, onDone: () {
        print("流已完成");
      }, cancelOnError: false);
    }

• Stream<T>. forEach()

  这中操作和listen()的方式基本差不多，也是一种监听流的方式，这只是监听了onData,下面代码也会输出0-9

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
      stream = stream.take(10);
      stream.forEach((data) {
        print(data);
      });
    }

• Stream<T> .length

  用于获取等待流中所有事件发射完成之后统计事件的总数量,下面代码会输出 10

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream.periodic(interval, (data) => data);
      stream = stream.take(10);
      var allEvents = await stream.length;
      print(allEvents);
    }

• Stream<T>.where

  在流中添加筛选条件，过滤掉一些不想要的数据，满足条件返回true,不满足条件返回false,如下我们筛选出流中大于5小于10的数据

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream<int>.periodic(interval, (data) => data);
      stream = stream.where((data)=>data>5);
      stream = stream.where((data)=> data<10);
      await for(int i in stream){
        print(i);
      }
    }

• stream<T>.map

  对流中的数据进行一些变换，以下是我对Stream的每个数据都加1

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream<int> stream = Stream<int>.periodic(interval, (data) => data);
      stream = stream.map((data) => data + 1);
      await for (int i in stream) {
        print(i);
      }
    }

• Stream<T>.expand

  对流中的数据进行一个扩展，如下，会输出1,1,2,2,3,3….

  void _stream() async {
      Duration interval = Duration(seconds: 1);
      Stream stream = Stream.periodic(interval, (data) => data);
      stream = stream.expand((data)=>[data,data]);
      stream.listen((data)=>print(data),onError:(error)=> print("发生错误") );
    }

• Stream<S>.transform

  如果我们在在流流转的过程中需要进行一些转换和控制我们则需要使用到transform,接收一个

  StreamTransformer<S,T>，S表示转换之前的类型，T表示转换后的输入类型，如下代码我们会接收到三组数字模拟输入了三次密码，并判断真确的密码，同时输出密码正确和密码错误：

  void _stream() async {
      var stream = Stream<int>.fromIterable([123456,234567,678901]);
      var st = StreamTransformer<int, String>.fromHandlers(
          handleData: (int data, sink) {
        if (data == 678901) {
          sink.add("密码输入正确,正在开锁。。。");
        } else {
          sink.add("密码输入错误...");
        }
      });
      stream.transform(st).listen((String data) => print(data),
          onError: (error) => print("发生错误"));
    }

  输入如下结果

  I/flutter (18980): 密码输入错误...
  I/flutter (18980): 密码输入错误...
  I/flutter (18980): 密码输入正确,正在开锁。。。

StreamController使用

​ 介绍完了Stream的基本概念和基本用法，上面直接创建流的方式，对我们本身开发来说，用途不是很大，我们在实际的开发过程中，基本都是使用的StreamContoller来创建流。通过源码我们可以知道Stream的几种构造方法，最终都是通过StreamController进行了包装。

创建StreamController对象及使用

• 构建单订阅的StreamController

//StreamController里面会创建一个Stream，我们实际操控的Stream
StreamController<String> streamController = StreamController();
streamController.stream.listen((data)=> print(data));
streamController.sink.add("aaa");
streamController.add("bbb");
streamController.add("ccc");
streamController.close();

//上面代码我们会输出 aaa,bbb,ccc

注意：如果我们给上面的代码再加一个listen会报如下异常,所以单订阅流，只能有一个listen。一般情况下我们多数都是使用的单订阅流，我们也可以将单订阅流转成多订阅流。

• 构建多监听器的StreamController有两种方式

  1.直接创建多订阅Stream

  StreamController<String> streamController = StreamController.broadcast();
      streamController.stream.listen((data){
        print(data);
      },onError: (error){
        print(error.toString());
      });
      streamController.stream.listen((data) => print(data));
      streamController.add("bbb");
  
      //上面代码回输出 bbb,bbb

  2.将单订阅流转成多订阅流

  StreamController<String> streamController = StreamController();
    Stream stream =streamController.stream.asBroadcastStream();
    stream.listen((data) => print(data));
    stream.listen((data) => print(data));
    streamController.sink.add("aaa");
    streamController.close();
    
    //上面代码会输出 aaa,aaa

注意：在流用完了之后记得关闭，调用streamController.close()

StreamBuilder使用

​ 前面我把Stream的常用方式做了简单的介绍和演示，我们怎么结合Flutter使用呢？在Flutter里面提供了一个Widget名叫StreamBuilder,StreamBuilder其实是个StreamBuilder它一直记录着流中最新的数据，当数据流发生变化时，会自动调用builder方法进行重建。

• StreamBuilder的源码如下，需要接受一个流，我们可以传入一个StreamController的Stream

  const StreamBuilder({
      Key key,
      this.initialData,
      Stream<T> stream,
      @required this.builder,
    }) : assert(builder != null),
         super(key: key, stream: stream);

• 使用StreamController 结合 StreamBuider对官方的计数器进行改进，取代setState刷新页面，代码如下

  class MyHomePage extends StatefulWidget {
    @override
    _MyHomePageState createState() => _MyHomePageState();
  }
  
  class _MyHomePageState extends State<MyHomePage> {
    int _count = 0;
    final StreamController<int> _streamController = StreamController();
  
    @override
    Widget build(BuildContext context) {
      return Scaffold(
        body: Container(
          child: Center(
            child: StreamBuilder<int>(
                stream: _streamController.stream,
                builder: (BuildContext context, AsyncSnapshot snapshot) {
                  return snapshot.data == null
                      ? Text("0")
                      : Text("${snapshot.data}");
                }),
          ),
        ),
        floatingActionButton: FloatingActionButton(
            child: const Icon(Icons.add),
            onPressed: () {
              _streamController.sink.add(++_count);
            }),
      );
    }
  
    @override
    void dispose() {
      _streamController.close();
      super.dispose();
    }
  }

源码相关

从StreamController源码我们可以看出来里面创建的得过程 默认创建一个_SyncStreamController,具体可以去读取下源码，看看StreamController是怎么创建Stream和StreamSink

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);
  }

注意：上面我既使用了streamController.sink.add("aaa");添加数据，也使用了streamController.add("bbb");方式添加数据。实际效果是一样的，查看源码可知 sink如下,实际上sink是对StreamController的一种包装，最终都是调取的StreamController.add方法；

class _StreamSinkWrapper<T> implements StreamSink<T> {
  final StreamController _target;
  _StreamSinkWrapper(this._target);
  void add(T data) {
    _target.add(data);
  }

  void addError(Object error, [StackTrace stackTrace]) {
    _target.addError(error, stackTrace);
  }

  Future close() => _target.close();

  Future addStream(Stream<T> source) => _target.addStream(source);

  Future get done => _target.done;

总结

    以上是我在开发过程中，对Flutter中Stream使用的一些简单的认识和常用方法的总结,StreamBuilder和 StreamController结合使用，实现局部刷新效果（比喻一个页面有多个接口，各自应该刷新自己控制的UI的部分而不应该整个页面刷新，时候我们可以使用StreamController和StreamBuilder达到这种效果）如有不到之处或者有偏差的地方，望指正~