Flutter响应式编程 - Stream

在Dart库中,有两种实现异步编程的方式(Future和Stream),使用它们只需要在代码中引入dart:async即可。

Flutter 中,整个 Stream 主要包含了 StreamController、Sink 、Stream 、StreamSubscription 四个对象,通过这四个对象来操控整个Stream的运行。

  • StreamController

有一个事件源叫 Stream,为了方便控制 Stream ,官方提供了使用 StreamController 作为管理;同时它对外提供了 StreamSink 对象作为事件输入口,可通过 sink 属性访问; 又提供 stream 属性提供 Stream 对象的监听和变换,最后得到的 StreamSubscription 可以管理事件的订阅。

可以说, StreamController就是如其名字所示一样用来管理其他三个对象的对象。

  • StreamSink

sink英文的意思为水槽,大家可以将其理解为日常生活中的厨房的洗碗槽,洗碗槽(sink)中的水(data)会流进管子(stream)中。一般作为事件的入口,提供如 add , addStream 等。

  • StreamSubscription

这是一个事件订阅后的对象,,Stream中有两种订阅模式,分别是单点订阅和广播。

表面上用于管理订阅过等各类操作,如 cacenl 、pause ,同时在内部也是事件的中转关键。

  • Stream

事件源本身,一般可用于监听事件或者对事件进行转换,如 listen 、 where 。

什么是Stream?

为了将Stream的概念可视化与简单化,可以将它想成是管道(pipe)的两端,它只允许从一端插入数据并通过管道从另外一端流出数据。
在Flutter中,

  • 大家将这样的管道称作Stream;
  • 为了控制Stream,大家通常可以使用StreamController来进行管理;
  • 为了向Stream中插入数据,StreamController提供了类型为StreamSink的属性sink作为入口;
  • StreamController提供stream属性作为数据的出口。

class StreamDemo extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('StreamDemo'),
        elevation: 0.0,
      ),
      body: StreamTestDemo(),
    );
  }
}

class StreamTestDemo extends StatefulWidget {
  @override
  State<StatefulWidget> createState() {
    // TODO: implement createState
    return StreamTestDemoState();
  }
}

class StreamTestDemoState extends State<StreamTestDemo> {
  StreamSubscription streamSubscription;
  StreamController<String> _streamDemo;
  StreamSink _sinkDemo;
  var _data = "--";

  @override
  void initState() {
    super.initState();
    //使用StreamController 管理Stream
    _streamDemo = StreamController<String>();
    //为了向Stream中插入数据,StreamController提供了类型为StreamSink的属性sink作为入口;
    _sinkDemo = _streamDemo.sink;
    //StreamSubscription 可以管理事件的订阅  取消 暫停  添加
    streamSubscription =
        _streamDemo.stream.listen(onData, onError: onError, onDone: onDone);
  }

  @override
  void dispose() {
    // TODO: implement dispose
    super.dispose();
    _streamDemo.close();
  }

  void onData(String data) {
    print('---------${data}');
    setState(() {
      _data = data;
    });
  }

  void onError(error) {
    print('--------error:$error');
  }

  void onDone() {
    print('--------Done');
  }

  Future<String> fetchData() async {
    await Future.delayed(new Duration(seconds: 3));
    // throw 'soming err';
    return 'hello word';
  }

  void _addDataToStream() async {
    print('------Add data to stream.');
    String data = await fetchData();
    // _streamDemo.add(data);
    _sinkDemo.add(data);
  }

  void _pauseStream() {
    print('------Pause subscription');
    streamSubscription.pause();
  }

  void _resumeStream() {
    print('------Resume subscription');
    streamSubscription.resume();
  }

  void _cancelStream() {
    print('------Cancel subscription');
    streamSubscription.cancel();
  }

  @override
  Widget build(BuildContext context) {
    // TODO: implement build
    return Container(
      child: Center(
        child: Column(
          mainAxisAlignment: MainAxisAlignment.center,
          children: <Widget>[
            Text(_data),
            Row(
              mainAxisAlignment: MainAxisAlignment.center,
              children: <Widget>[
                FlatButton(
                  child: Text('Add'),
                  onPressed: _addDataToStream,
                ),
                FlatButton(
                  child: Text('Pause'),
                  onPressed: _pauseStream,
                ),
                FlatButton(
                  child: Text('Resume'),
                  onPressed: _resumeStream,
                ),
                FlatButton(
                  child: Text('Cancel'),
                  onPressed: _cancelStream,
                ),
              ],
            ),
          ],
        ),
      ),
    );
  }
}

如下圖 点击add 后3s后会显示hello word 同时也可以点击 暂停 恢复 取消等 查看结果


image.png

另外 Stream 还支撑多次订阅
修改代码

  @override
  void initState() {
    super.initState();
    print('-----Create a stream.');
    //支撑多次订阅
    _streamDemo = StreamController.broadcast();
    _sinkDemo = _streamDemo.sink;
    print('-----Start listening on a stream.');
    _streamDemoSubscription =
        _streamDemo.stream.listen(onData, onError: onError, onDone: onDone);
    _streamDemo.stream.listen(onDataTwo, onError: onError, onDone: onDone);
    print('-----Initialize completed.');
  }


  void onDataTwo(String data) {
    print('-----onDataTwo: $data');
  }

运行后便可以查看到两次的ondata 输出