Flutter 详解 （六、深入了解Stream）

Future

Future有三种状态未完成、完成带有值、完成带有异常，使用Future可以简化事件任务。 假如你有一个按钮，点击之后开始下载图片，首先事件循环机制会处理你的点击事件，然后开始下载图片，当下载完成，你可以使用then来注册回调，然后获取到图片并显示出来。

通常我们不会直接创建，网络下载图片会返回一个Future,文件I/O会返回一个Future,那我们怎么创建一个呢？只需要关键字async就表示该函数异步执行，返回类型是Future<T>。

Future<String> getStr()async{
  var str = HttpRequest.getString('www.fgyong.cn');
  return str;
}

使用http请求地址www.fgyong.cn获取数据，然后返回。

如何接收文本呢？

其实很简单，只需要使用await关键字即可，用来注册then回调。

main(List<String> args) async {
  String string = await getStr();
  print(string);
}

等同于：

main(List<String> args) async {
  getStr().then((value) {
    print(value);
  });
}

官方比较推荐前者，因为前者看起来很像同步函数，少了层层嵌套，方便开发者理解代码。
网络下载想延迟动画隐藏时间，可以使用Future.delayed()。

await Future.delayed(Duration(seconds: 2), () {
hideAnimation();
});

如果已经带有值的想异步去执行，那么可以使用Future.value()

Stream

dart：async库包含对许多Dart API很重要的两种类型：Stream和Future。如果Future表示单个计算的结果，则流是一系列结果。您侦听流以获取有关结果（数据和错误）以及流关闭的通知。您还可以在收听流时暂停播放或在流完成之前停止收听。

如何使用Stream

流可以通过多种方式创建，后续在仔细讲解，但是它们都可以以相同的方式使用：异步for循环（通常仅称为await for）遍历流的事件，如for循环迭代遍历。例如：

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}

此代码仅接收整数事件流中的每个事件，将它们相加，然后返回（和）其和。当循环主体结束时，函数将暂停，直到下一个事件到达或流完成为止。 该函数标记有async关键字，在使用await for循环时需要此关键字。 以下示例通过使用async *函数生成简单的整数流来测试前面的代码：

import 'dart:async';

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}

Stream<int> countStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield i;
  }
}

main() async {
  var stream = countStream(10);
  var sum = await sumStream(stream);
  print(sum); // 55
}

当流中没有更多事件时，就完成流，并通知接收事件的代码，就像通知新事件到来一样。使用await for循环读取事件时，流完成后循环停止。 在某些情况下，流完成之前会发生错误；可能是网络从远程服务器上获取文件时发生故障，或者创建事件的代码存在错误，但是有人需要了解它。 流还可以传递错误事件，就像传递数据事件一样。大多数流将在出现第一个错误后停止，但有可能传递多个错误的流以及在发生错误事件后传递更多数据的流。在本文档中，我们仅讨论最多产生一个错误的流。 当使用await for读取流时，循环语句会引发错误。这也结束了循环。您可以使用try-catch捕获错误。以下示例在循环迭代器等于时引发错误 4：

import 'dart:async';

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  try {
    await for (var value in stream) {
      sum += value;
    }
  } catch (e) {
    return -1;
  }
  return sum;
}

Stream<int> countStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    if (i == 4) {
      throw new Exception('Intentional exception');
    } else {
      yield i;
    }
  }
}

main() async {
  var stream = countStream(10);
  var sum = await sumStream(stream);
  print(sum); // -1
}

两种Stream

单一订阅流 最常见的流包含一系列事件，这些事件是较大整体的一部分。事件必须以正确的顺序传递，并且不能丢失任何事件。这是您在读取文件或接收Web请求时获得的流。 这样的流只能被收听一次。稍后再次收听可能意味着错过了最初的事件，然后其余部分毫无意义。当您开始收听时，数据将被提取并以块的形式提供。 广播流 另一种流是针对可以一次处理的单个消息的。例如，这种流可用于浏览器中的鼠标事件。 您可以随时开始收听这样的流，并且在收听时会触发事件。多个收听者可以同时收听，并且您可以在取消上一个订阅之后稍后再次收听。

Stream的方法

Future<T> get first;
Future<bool> get isEmpty;
Future<T> get last;
Future<int> get length;
Future<T> get single;
Future<bool> any(bool Function(T element) test);
Future<bool> contains(Object needle);
Future<E> drain<E>([E futureValue]);
Future<T> elementAt(int index);
Future<bool> every(bool Function(T element) test);
Future<T> firstWhere(bool Function(T element) test, {T Function() orElse});
Future<S> fold<S>(S initialValue, S Function(S previous, T element) combine);
Future forEach(void Function(T element) action);
Future<String> join([String separator = ""]);
Future<T> lastWhere(bool Function(T element) test, {T Function() orElse});
Future pipe(StreamConsumer<T> streamConsumer);
Future<T> reduce(T Function(T previous, T element) combine);
Future<T> singleWhere(bool Function(T element) test, {T Function() orElse});
Future<List<T>> toList();
Future<Set<T>> toSet();

这么多方法基本都可以使用await for来循环

Future<bool> contains(Object needle) async {
  await for (var event in this) {
    if (event == needle) return true;
  }
  return false;
}

Future forEach(void Function(T element) action) async {
  await for (var event in this) {
    action(event);
  }
}

Future<List<T>> toList() async {
  final result = <T>[];
  await this.forEach(result.add);
  return result;
}

Future<String> join([String separator = ""]) async =>
    (await this.toList()).join(separator);

修改Stream

Stream<R> cast<R>();
Stream<S> expand<S>(Iterable<S> Function(T element) convert);
Stream<S> map<S>(S Function(T event) convert);
Stream<T> skip(int count);
Stream<T> skipWhile(bool Function(T element) test);
Stream<T> take(int count);
Stream<T> takeWhile(bool Function(T element) test);
Stream<T> where(bool Function(T event) test);

这些基本都是使用闭包过滤流的内容，当然也可以转换内容。

监听Stream

最后是监听，当流改变时会触发监听listen()，所有的流都可以被监听。

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

创建自己的Stream

创建Stream大概有三种，如下所示：

  • 转换Stram
  • 使用async*创建Stream
  • 使用StreamController创建

转换Stram

经常有些Stream包含的值不是我们想要的，那么就需要我们转换一下了，例如我们把数字转成字符串.

  Future<String> _toString() async {
    var s = await _stream().map((event) => event.toString()).join('|');
    return s;
  }

也可以转成数组

 Future<List> _toList() async {
    var s = await _stream().toList();
    return s;
  }

使用async*创建Stream

创建新流的一种方法是使用异步生成器（async *）函数。在调用该函数时创建该流，并且在侦听该流时该函数的主体开始运行。函数返回时，流关闭。在函数返回之前，它可以使用yield或yield *语句在流上发出事件。

这是一个原始示例，该示例会定期发射数字：

Stream<int> timedCounter(Duration interval, [int maxCount]) async* {
  int i = 0;
  while (true) {
    await Future.delayed(interval);
    yield i++;
    if (i == maxCount) break;
  }
}

此函数返回一个Stream。当收听该流时，主体开始运行。它反复延迟请求的时间间隔，然后产生下一个数字。如果忽略count参数，则循环上没有停止条件，因此流永远输出越来越大的数字-或直到侦听器取消其订阅为止。 当侦听器取消时（通过在listen（）方法返回的StreamSubscription对象上调用cancel（）），则主体下一次到达yield语句时，yield将充当return语句。执行所有封闭的finally块，然后函数退出。如果函数尝试在退出前产生一个值，则该操作将失败并充当返回值。 当函数最终退出时，由cancel（）方法返回的Future完成。如果函数以错误退出，则Future会以该错误结束；否则，它以null结束。 另一个更有用的示例是一个转换序列的函数：

Stream<T> streamFromFutures<T>(Iterable<Future<T>> futures) async* {
  for (var future in futures) {
    var result = await future;
    yield result;
  }
}

使用yield*

使用yield*来调用其他的函数取下一个值，当不获取的时候，则不运行。

  Stream<int> _stream() async* {
    if (_count < 10) {
      yield _count++;
      await Future.delayed(Duration(seconds: 1));
      sleep(Duration(seconds: 1));
      yield* _getDataFromServer();
    }
  }

yield*和yield区别是后者直接返回一个固定的值，而前者返回是的是一个函数。前者多用于分流，把一个Stream<T>分为其他的Stream<R>或Stream<E>。

使用StreamController创建stream 

_streamController = StreamController();
// 监听
_streamController.stream.listen((event) { })
/// 添加数据
_streamController.add('data');

StreamControlelr的类图如下所示，他们从Sink到StreamSink都只是到导入接口，并无直接继承关系，在抽象类方面按照功能解耦，最终由StreamController整合，负责添加的是StreamControlelr,负责监听的是Stream,最终在当前的Zone中执行回调Zone.runUnaryGuarded()。Zone类似一个沙盒环境，在APP启动的时候创建。

image

原理

StreamController中初始化，直接调用了_SyncStreamController，具体功能全部在_SyncStreamController(同步)或_AsyncStreamController(异步)中实现。

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

最终在_BufferingStreamSubscription类中，实现了
onDone和onError函数，他们分别在类实例化的时候注册回调函数。

void onData(void handleData(T event)) {
  handleData ??= _nullDataHandler;//默认值
  _onData = _zone.registerUnaryCallback<dynamic, T>(handleData);
}
void onDone(void handleDone()) {
  handleDone ??= _nullDoneHandler;
  _onDone = _zone.registerCallback(handleDone);
}

那么listener函数是在何时添加的呢？
在获取stream是_ControllerStream(),最终listener函数是_StreamImpl extends Steam中实现的，代码如下：

StreamSubscription<T> listen(void onData(T data),
      {Function onError, void onDone(), bool cancelOnError}) {
    cancelOnError = identical(true, cancelOnError);
    StreamSubscription<T> subscription =
        _createSubscription(onData, onError, onDone, cancelOnError);
    _onListen(subscription);
    return subscription;
  }
  
/// 创建一个订阅者
StreamSubscription<T> _createSubscription(void onData(T data),
    Function onError, void onDone(), bool cancelOnError) {
  return new _BufferingStreamSubscription<T>(
      onData, onError, onDone, cancelOnError);
}

最终监听是执行的_BufferingStreamSubscription的_onData(void HandleData(T event))，这里是使用了_zone.registerUnaryCallback来注册回调函数，否则在调用的时候会报错。

void onData(void handleData(T event)) {
  handleData ??= _nullDataHandler;
  _onData = _zone.registerUnaryCallback<dynamic, T>(handleData);
}

在stream.add()函数执行了_add(),源码如下

void _add(T data) {
  assert(!_isClosed);
  if (_isCanceled) return;
  if (_canFire) {
    _sendData(data);
  } else {
    _addPending(new _DelayedData<T>(data));
  }
}

当状态已关闭，直接断言，当状态已取消，返回操作，当可以发送数据，则执行_sendData()函数，该函数才是最终发送数据，执行listen操作的关键函数，源码如下：

void _sendData(T data) {
  assert(!_isCanceled);
  assert(!_isPaused);
  assert(!_inCallback);
  bool wasInputPaused = _isInputPaused;
  _state |= _STATE_IN_CALLBACK;//取出来第六位
  _zone.runUnaryGuarded(_onData, data);
  _state &= ~_STATE_IN_CALLBACK;// 舍弃第六位
  _checkState(wasInputPaused);
}

通过final Zone _zone = Zone.current;获取当前的Zone来执行已经注册的回调_zone.runUnaryGuarded(_onData, data),执行完毕,使用_state&=~_STATE_IN_CALLBACK来保存当前状态，使用&=~舍弃第6位数字，_STATE_IN_CALLBACK值为32，那么低第六位是1，~_STATE_IN_CALLBACK,除了第六位，剩下的都是1，然后&=来取出来其他的位数值保存下来。

然后在_checkState(wasInutPaused)来确定在执行callback中间并无状态改变。

那么再执行完毕_sendDone()的时候也是如此，首先判断是否已经取消，没取消的话，执行_onDone回调。

 void _sendDone() {
    assert(!_isCanceled);
    assert(!_isPaused);
    assert(!_inCallback);

    void sendDone() {
      // If the subscription has been canceled while waiting for the cancel
      // future to finish we must not report the done event.
      if (!_waitsForCancel) return;
      _state |= (_STATE_CANCELED | _STATE_CLOSED | _STATE_IN_CALLBACK);
      _zone.runGuarded(_onDone);
      _state &= ~_STATE_IN_CALLBACK;
    }

    _cancel();
    _state |= _STATE_WAIT_FOR_CANCEL;
    if (_cancelFuture != null &&
        !identical(_cancelFuture, Future._nullFuture)) {
      _cancelFuture.whenComplete(sendDone);
    } else {
      sendDone();
    }
  }

事件流就是在前期使用Zone注册，在add的时候使用Zone调用已经注册好的回调，广播是循环调用

关键函数：

/// 注册回调
_zone.registerUnaryCallback<dynamic, T>(handleData)

/// 注册不带参数的回调

_zone.registerCallback(R callback())

/// 执行 附带参数的回调

_zone.runUnaryGuarded(_onData, data)

更多API可以查看官方源码。

参考

文章汇总

<<Dart 异步与多线程>>

<<Flutter 详解（一、深入了解状态管理--ScopeModel）>>

<<Flutter 详解(二、深入了解状态管理--Redux)>>

<<Flutter 详解（三、深入了解状态管理--Provider)>>

<<Flutter 详解（四、深入了解状态管理--BLoC)>>

<<Flutter 详解 (五、深入了解--Key)>>

<<Flutter 详解 （六、深入了解--Stream>>

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,126评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,254评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,445评论 0赞 341
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,185评论 1赞 278
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,178评论 5赞 371
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,970评论 1赞 284
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,276评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,927评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,400评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,883评论 2赞 323
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,997评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,646评论 4赞 322
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,213评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,204评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,423评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,423评论 2赞 352
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,722评论 2赞 345