AI 助理
备案控制台
开发者社区 数据库 文章 正文

Flutter 80: 初识 Flutter Stream (一)

2020-03-22 1044
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 0 基础学习 Flutter，第八十步：学习一下基础 Flutter Stream！

      小菜在之前尝试 EventChannel 时曾经用到过 Stream 流数据，现在准备学习一下 flutter_bloc 时也主要用到 Stream 来做异步处理，于是简单学习一下何为 Stream？

A source of asynchronous data events.

      Stream 主要应用于 Flutter 的异步操作，在其他编程语言中也存在；Stream 提供了一种接受事件队列的方法，可通过 listen 进行数据监听，通过 error 接收失败状态，通过 done 来接收结束状态；

1. Stream 创建

      Flutter 提供了多种创建 Stream 的方式；

1.1 Stream.fromFuture(Future future)

      Stream 通过 Future 创建新的单订阅流，当 Future 完成时会触发 data / error，然后以 done 事件结束；

Future<String> getData() async {
  await Future.delayed(Duration(seconds: 3));
  return '当前时间为：${DateTime.now()}';
}

_streamFromFuture() {
  Stream.fromFuture(getData())
      .listen((event) => print('Stream.fromFuture -> $event'))
      .onDone(() => print('Stream.fromFuture -> done 结束'));
}

1.2 Stream.fromFutures(Iterable> futures)

      Stream 通过一系列的 Future 创建新的单订阅流，每个 Future 都会有自身的 data / error 事件，当这一系列的 Future 均完成时，Stream 以 done 事件结束；若 Futures 为空，则 Stream 会立刻关闭；其分析源码，很直接的看到是将每一个 Future 事件监听完之后才会执行的微事件结束；

====================================== 源码 ======================================
factory Stream.fromFutures(Iterable<Future<T>> futures) {
    _StreamController<T> controller =
        new _SyncStreamController<T>(null, null, null, null);
    int count = 0;
    var onValue = (T value) {
      if (!controller.isClosed) {
        controller._add(value);
        if (--count == 0) controller._closeUnchecked();
      }
    };
    var onError = (error, StackTrace stack) {
      if (!controller.isClosed) {
        controller._addError(error, stack);
        if (--count == 0) controller._closeUnchecked();
      }
    };
    for (var future in futures) {
      count++;
      future.then(onValue, onError: onError);
    }
    // Use schedule microtask since controller is sync.
    if (count == 0) scheduleMicrotask(controller.close);
    return controller.stream;
}
====================================== 测试 ======================================
_streamFromFutures() {
  var data = [getData(), getData(), getData()];
  Stream.fromFutures(data)
      .listen((event) => print('Stream.fromFutures -> $event'))
      .onDone(() => print('Stream.fromFutures -> done 结束'));
}

1.3 Stream.fromIterable(Iterable elements)

      Stream 通过数据集合中获取并创建单订阅流，通过 listen 监听迭代器中每一个子 element，当 Stream 监听到取消订阅或 Iterator.moveNext 返回 false / throw 异常 时停止迭代；

_streamFromIterable() {
  var data = [1, 2, '3.toString()', true, false, 6];
  Stream.fromIterable(data)
      .listen((event) => print('Stream.fromIterable -> $event'))
      .onDone(() => print('Stream.fromIterable -> done 结束'));
}

1.4 Stream.periodic(Duration period, [T computation(int computationCount)])

      Stream 通过 Duration 对象作为参数创建一个周期性事件流，其中若不设置 computation 时 onData 获取数据为 null；若没有事件结束则会一直周期性执行；

_streamFromPeriodic() {
  Duration interval = Duration(seconds: 1);
  // onData 获取数据为 null
  Stream<int> stream = Stream<int>.periodic(interval);
  stream.listen((event) {
  print('Stream.periodic -> $event');
  }).onDone(() {
  print('Stream.periodic -> done 结束');
  });

  // onData 获取数据为 int 类型 data
  Stream<int> streamData = Stream<int>.periodic(interval, (data) => data);
  streamData.listen((event) {
    print('Stream.periodic -> $event');
    if (event >= 10) {}
  }).onDone(() {
    print('Stream.periodic -> done 结束');
  });
}


2. Stream 基本操作

2.1 Stream take(int count)

      take() 对于单订阅方式，可以提供 take 设置之前的 Stream 订阅数据，例如设置中断 Stream.periodic 周期展示次数；小菜粗略理解为 take 可以作为中断订阅，如果 take 设置次数大于 onDone 之前的订阅数据次数，Stream 依旧获取所有 onDone 之前的订阅数据；

_streamFromPeriodic() {
  Duration interval = Duration(seconds: 1);
  Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
  streamData.take(5).listen((event) {
    print('Stream.periodic -> $event');
  }).onDone(() {
    print('Stream.periodic -> done 结束');
  });
}

_streamFromIterable() {
  var data = [1, 2, '3.toString()', true, false, 6];
  Stream.fromIterable(data)
      .take(8)
      .listen((event) => print('Stream.fromIterable -> $event'))
      .onDone(() => print('Stream.fromIterable -> done 结束'));
}

2.2 Stream takeWhile(bool test(T element))

      takeWhile 也可以实现上述相同效果，通过 test 返回一个 boolean 类型，如果为 false 则中断订阅；

_streamFromPeriodic() {
  Duration interval = Duration(seconds: 1);
  Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
  streamData.takeWhile((element) {
    print('Stream.periodic.takeWhile -> $element');
    return element <= 5;
  }).listen((event) {
    print('Stream.periodic -> $event');
  }).onDone(() {
    print('Stream.periodic -> done 结束');
  });
}

2.3 Stream where(bool test(T event))

      where 用于在当前 Stream 中创建一个新的 Stream 用来丢弃不符合 test 的数据；小菜简单理解为类似数据库查询一样，仅过滤符合需求的数据流；且 where 可以设置多次；

Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
streamData.takeWhile((element) {
  print('Stream.periodic.takeWhile -> $element');
  return element <= 5;
}).where((event) {
  print('Stream.periodic.where -> $event');
  return event > 3;
}).listen((event) {
  print('Stream.periodic -> $event');
}).onDone(() {
  print('Stream.periodic -> done 结束');
});

2.4 Stream distinct([bool equals(T previous, T next)])

      distinct 小菜理解为相邻两个数据滤重；

var data = [1, 2, '3.toString()', true, true, false, true, 6];
Stream.fromIterable(data)
    .distinct()
    .listen((event) => print('Stream.fromIterable -> $event'))
    .onDone(() => print('Stream.fromIterable -> done 结束'));

2.5 Stream skip(int count)

      skip 用于跳过符合条件的订阅数据次数；

Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
streamData.takeWhile((element) {
      print('Stream.periodic.takeWhile -> $element');
      return element <= 6;
    }).where((event) {
      print('Stream.periodic.where -> $event');
      return event > 2;
    })
    .skip(2).listen((event) {
      print('Stream.periodic -> $event');
    }).onDone(() {
      print('Stream.periodic -> done 结束');
    });

2.6 Stream skipWhile(bool test(T element))

      skipWhile 用于跳过在 where 符合条件下满足设置 test 条件的订阅数据；即当 test 为 true 时跳过当前订阅数据监听；

Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
streamData.takeWhile((element) {
  print('Stream.periodic.takeWhile -> $element');
  return element <= 6;
}).where((event) {
  print('Stream.periodic.where -> $event');
  return event > 2;
}).skipWhile((element) {
  print('Stream.periodic.skipWhile -> $element');
  return element <= 4;
}).listen((event) {
  print('Stream.periodic -> $event');
}).onDone(() {
  print('Stream.periodic -> done 结束');
});

2.7 Stream map(S convert(T event))

      在当前 Stream 基础上创建一个新的 Stream 并对当前 Stream 进行数据操作，onData 监听到的是 map 变更后的新的数据流；

Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
streamData.takeWhile((element) {
  print('Stream.periodic.takeWhile -> $element');
  return element <= 6;
}).where((event) {
  print('Stream.periodic.where -> $event');
  return event > 2;
}).skipWhile((element) {
  print('Stream.periodic.skipWhile -> $element');
  return element <= 4;
}).map((event) {
  print('Stream.periodic.map -> $event -> ${event * 100}');
  return event * 100;
}).listen((event) {
  print('Stream.periodic -> $event');
}).onDone(() {
  print('Stream.periodic -> done 结束');
});

2.8 Stream expand(Iterable convert(T element))

      在当前 Stream 基础上创建新的 Stream 并将当前订阅数据转为新的订阅数据组，onData 监听 数据组 中每个新的订阅数据元素；

Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
streamData.takeWhile((element) {
  print('Stream.periodic.takeWhile -> $element');
  return element <= 6;
}).where((event) {
  print('Stream.periodic.where -> $event');
  return event > 2;
}).skipWhile((element) {
  print('Stream.periodic.skipWhile -> $element');
  return element <= 4;
}).expand((element) {
  print('Stream.periodic.expand -> $element -> ${element * 10} -> ${element * 100}');
  return [element, element * 10, element * 100];
}).listen((event) {
  print('Stream.periodic -> $event');
}).onDone(() {
  print('Stream.periodic -> done 结束');
});

2.9 Future get length

      Stream 监听订阅事件结束后，符合 where 条件的数量；

_streamLength(index) async {
  Duration interval = Duration(seconds: 1);
  Stream<int> streamData = Stream<int>.periodic(interval, (data) => data + 1);
  streamData = streamData.takeWhile((element) {
    print('Stream.periodic.takeWhile -> $element');
    return element <= 6;
  }).where((event) {
    print('Stream.periodic.where -> $event');
    return event > 2;
  }).skipWhile((element) {
    print('Stream.periodic.skipWhile -> $element');
    return element <= 4;
  });
  switch (index) {
    case 1:
      var length = await streamData.length;
      print('Stream.length -> $length');
      break;
    case 2:
      var isEmpty = await streamData.isEmpty;
      print('Stream.isEmpty -> $isEmpty');
      break;
    case 3:
      var isBroadcast = await streamData.isBroadcast;
      print('Stream.isBroadcast -> $isBroadcast');
      break;
    case 4:
      var first = await streamData.first;
      print('Stream.first -> $first');
      break;
    case 5:
      var last = await streamData.last;
      print('Stream.last -> $last');
      break;
  }
}

2.10 Future get isEmpty

      Stream 监听订阅事件结束后，统计是否符合 where 条件的订阅数据是否为空；

_streamLength(2);

2.11 Future get first

      获取 Stream 符合条件的第一个订阅数据；

_streamLength(4);

2.12 Future get last

      获取 Stream 符合条件的最后一个订阅数据；

_streamLength(5);

2.13 Future> toList()

      在 Stream 监听结束之后，将订阅数据存储在 List 中，该操作为异步操作；

_streamToList() async {
  var data = [1, 2, '3.toString()', true, true, false, true, 6];
  Stream stream = Stream.fromIterable(data).distinct();
  List list = await stream.toList();
  if (list != null) {
    print('Stream.toList -> ${list}');
    for (int i = 0; i < list.length; i++) {
      print('Stream.toList -> ${i + 1} -> ${list[i]}');
    }
  }
}

2.14 Future> toSet()

      在 Stream 监听结束之后，将订阅数据存储在 Set 中，Set 可以过滤重复数据；

_streamToSet() async {
  var data = [1, 2, '3.toString()', true, true, false, true, 6];
  Stream stream = Stream.fromIterable(data);
  Set set = await stream.toSet();
  if (set != null) {
    print('Stream.toSet -> ${set}');
  }
}

2.15 Future forEach(void action(T element))

      监听 Stream 中订阅数据，是对 listen 方式的一种监听；

_streamForEach() {
  var data = [1, 2, '3.toString()', true, true, false, true, 6];
  Stream stream = Stream.fromIterable(data).distinct();
  stream.forEach((element) => print('Stream.forEach -> $element'));
}

      小菜对 Stream 的尝试才刚刚开始，还有众多方法未曾尝试，对 Stream 的理解还很浅显，如有错误请多多指导！

来源： 阿策小和尚

文章标签：
数据库
存储
关键词：
Flutter Stream
Flutter初识Stream
阿策～
目录
相关文章
旺仔大牛
|
3月前
|
Dart 前端开发 JavaScript
Flutter&Dart-异步编程Future、Stream极速入门
Flutter&Dart-异步编程Future、Stream极速入门
旺仔大牛
72 4
Flutter&Dart-异步编程Future、Stream极速入门
我是坚果
Flutter：Stream.periodic 示例
本文将带您了解在 Flutter中使用Stream.periodic的完整示例 该Stream.periodic构造，顾名思义，是用来创建流，在周期间隔反复广播事件。简单用法：
我是坚果
226 0
阿策～
Flutter 82: 初识 Flutter Stream (二)
0 基础学习 Flutter，第八十二步：学习基础的 Stream 和 StreamController！
阿策～
1027 0
游客zi3qmblbcdexu
|
1月前
|
Android开发 iOS开发 容器
鸿蒙harmonyos next flutter混合开发之开发FFI plugin
鸿蒙harmonyos next flutter混合开发之开发FFI plugin
游客zi3qmblbcdexu
116 2
西瓜味的敌敌畏
|
5月前
|
开发框架 前端开发 测试技术
Flutter开发常见问题解答
Flutter开发常见问题解答
西瓜味的敌敌畏
84 5
zacksleo
|
28天前
|
开发者
鸿蒙Flutter实战：07-混合开发
鸿蒙Flutter混合开发支持两种模式：1) 基于har包，便于主项目开发者无需关心Flutter细节，但不支持热重载；2) 基于源码依赖，利于代码维护与热重载，需配置Flutter环境。项目结构包括AppScope、flutter_module等目录，适用于不同开发需求。
zacksleo
69 3
lqj_本人
|
13天前
|
传感器 开发框架 物联网
鸿蒙next选择 Flutter 开发跨平台应用的原因
鸿蒙（HarmonyOS）是华为推出的一款旨在实现多设备无缝连接的操作系统。为了实现这一目标，鸿蒙选择了 Flutter 作为主要的跨平台应用开发框架。Flutter 的跨平台能力、高性能、丰富的生态支持和与鸿蒙系统的良好兼容性，使其成为理想的选择。通过 Flutter，开发者可以高效地构建和部署多平台应用，推动鸿蒙生态的快速发展。
lqj_本人
114 0
lqj_本人
|
15天前
|
Dart 安全 UED
Flutter&鸿蒙next中的表单封装：提升开发效率与用户体验
在移动应用开发中，表单是用户与应用交互的重要界面。本文介绍了如何在Flutter中封装表单，以提升开发效率和用户体验。通过代码复用、集中管理和一致性的优势，封装表单组件可以简化开发流程。文章详细讲解了Flutter表单的基础、封装方法和表单验证技巧，帮助开发者构建健壮且用户友好的应用。
lqj_本人
55 0
shuj
|
1月前
|
开发框架 移动开发 Android开发
安卓与iOS开发中的跨平台解决方案：Flutter入门
【9月更文挑战第30天】在移动应用开发的广阔舞台上，安卓和iOS两大操作系统各自占据半壁江山。开发者们常常面临着选择：是专注于单一平台深耕细作，还是寻找一种能够横跨两大系统的开发方案？Flutter，作为一种新兴的跨平台UI工具包，正以其现代、响应式的特点赢得开发者的青睐。本文将带你一探究竟，从Flutter的基础概念到实战应用，深入浅出地介绍这一技术的魅力所在。
shuj
74 7
zacksleo
|
28天前
|
编解码 Dart API
鸿蒙Flutter实战：06-使用ArkTs开发Flutter鸿蒙插件
本文介绍了如何开发一个 Flutter 鸿蒙插件，实现 Flutter 与鸿蒙的混合开发及双端消息通信。通过定义 `MethodChannel` 实现 Flutter 侧的 token 存取方法，并在鸿蒙侧编写 `EntryAbility` 和 `ForestPlugin`，使用鸿蒙的首选项 API 完成数据的读写操作。文章还提供了注意事项和参考资料，帮助开发者更好地理解和实现这一过程。
zacksleo
56 0