说明
Stream 也是用于接收异步事件数据，和 Future 不同的是，它可以接收多个异步操作的结果（成功或失败）。 也就是说，在执行异步任务时，可以通过多次触发成功或失败事件来传递结果数据或错误异常。简单理解，其实就是一个异步数据队列而已。我们知道队列的特点是先进先出的，Stream也正是如此。 Stream 常用于会多次读取数据的异步任务场景，如网络内容下载、文件读写等。在Dart语言中，Stream有两种类型，一种是点对点的单订阅流（Single-subscription），另一种则是广播流。
stream几个构造方法：
- Stream.periodic:构造一个周期流，有2个参数第一个参数为间隔时间，第二个参数为回调函数。
- Stream.fromFuture:从Future创建新的单订阅流,当future完成时将触发一个data或者error，然后使用Down事件关闭这个流。
- Stream.fromFutures:从一组Future创建一个单订阅流，每个future都有自己的data或者error事件，当整个Futures完成后，流将会关闭。如果Futures为空，流将会立刻关闭。
- Stream.fromIterable:创建从一个集合中获取其数据的单订阅流。
- Stream.value：用于从单个值创建Stream。
void main(){
test();
}
test() async{
// 1,使用 periodic 创建流，第一个参数为间隔时间，第二个参数为回调函数
Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
// 2,从Future创建Stream
Future<String> fut = Future((){
return "async task";
});
Stream<String> stream = Stream<String>.fromFuture(fut);
//3, 将多个Future放入一个列表中，将该列表传入
Future<String> fut1 = Future((){
// 模拟耗时5秒
sleep(Duration(seconds:5));
return "async task1";
});
Future<String> fut2 = Future((){
return "async task2";
});
Stream<String> stream = Stream<String>.fromFutures([fut1,fut2]);
// 4,从一个列表创建`Stream`
Stream<int> stream = Stream<int>.fromIterable([1,2,3]);
// 5,value
Stream<bool> stream = Stream<bool>.value(false);
// await for循环从流中读取
await for(var i in stream){
print(i);
}
}
// 可以在回调函数中对值进行处理，这里直接返回了
int callback(int value){
return value;
}
监听 Stream
监听Stream，并从中获取数据也有三种方式:
- 一种就是我们上文中使用的await for循环，这也是官方推荐的方式，看起来更简洁友好；
- 使用forEach方法；
- 使用listen方法StreamSubscription<T> listen(void onData(T event), {Function onError, void onDone(), bool cancelOnError})。
onData(必填)：收到数据时触发；onError：收到Error时触发；onDone：结束时触发；unsubscribeOnError：遇到第一个Error时是否取消订阅，默认为false。
test() async{
Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), callback);
//1, await for循环从流中读取
await for(var i in stream){
print(i);
}
//2, 使用forEach，传入一个函数进去获取并处理数据
stream.forEach((int x){
print(x);
});
//3,listen
stream.listen(
(x)=>print(x),
onError: (e)=>print(e),
onDone: ()=>print("onDone"));
}
Stream的一些方法
take 和 takeWhile
Stream<T> take(int count) 用于限制Stream中的元素数量;
Stream<T>.takeWhile(bool test(T element)) 与 take作用相似，只是它的参数是一个函数类型，且返回值必须是一个bool值.
skip 和 skipWhile
Stream<T> skip (int count)，跳过元素。请注意，该方法只是从Stream中获取元素时跳过，被跳过的元素依然是被执行了的，所耗费的时间依然存在，其实只是跳过了执行完的结果而已。
Stream<T> skipWhile(bool test(T element)) 方法与takeWhile用法是相同的，传入一个函数对结果进行判断，表示跳过满足条件的
toList
Future<List<T>> toList() 表示将Stream中所有数据存储在List中
属性 length
stream.length 等待并获取流中所有数据的数量
Stream 2个辅助类
StreamController
Stream的一个帮助类，可用于整个 Stream 过程的控制。使用该类时，需要导入'dart:async'，其add方法和sink.add方法是相同的，都是用于放入一个元素，addError方法用于产生一个错误，监听方法中的onError可获取错误。
import 'dart:async';
void main() {
test();
}
test() async{
// 创建
StreamController streamController = StreamController();
//也可以传入一个指定的stream sc.addStream(stream);
// 放入事件
streamController.add('element_1');
streamController.addError("this is error");
streamController.sink.add('element_2');
streamController.stream.listen(
print,
onError: print,
onDone: ()=>print("onDone"));
}
StreamController的原型，它有5个可选参数：factory StreamController( {void onListen(),void onPause(),void onResume(),onCancel(),bool sync: false})，可以调用对于的方法控制流，同时回调处理对于的节点。
- onListen 注册监听时回调
- onPause 当流暂停时回调
- onResume 当流恢复时回调
- onCancel 当监听器被取消时回调
- sync 当值为true时表示同步控制器SynchronousStreamController，默认值为false，表示异步控制器
StreamTransformer
该类可以使我们在Stream上执行数据转换。然后，这些转换被推回到流中，以便该流注册的所有监听器可以接收；
构造方法：factory StreamTransformer.fromHandlers({ void handleData(S data, EventSink<T> sink),
void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
void handleDone(EventSink<T> sink)})
- handleData：响应从流中发出的任何数据事件。提供的参数是来自发出事件的数据，以及EventSink<T>，表示正在进行此转换的当前流的实例
- handleError：响应从流中发出的任何错误事件
- handleDone：当流不再有数据要处理时调用。通常在流的close()方法被调用时回调
void test() {
StreamController sc = StreamController<int>();
// 创建 StreamTransformer对象
StreamTransformer stf = StreamTransformer<int, double>.fromHandlers(
handleData: (int data, EventSink sink) {
// 操作数据后，转换为 double 类型
sink.add((data * 2).toDouble());
},
handleError: (error, stacktrace, sink) {
sink.addError('wrong: $error');
},
handleDone: (sink) {
sink.close();
},
);
// 调用流的transform方法，传入转换对象
Stream stream = sc.stream.transform(stf);
stream.listen(print);
// 添加数据，这里的类型是int
sc.add(1);
sc.add(2);
sc.add(3);
// 调用后，触发handleDone回调
// sc.close();
}
//输出结果：
//2.0
//4.0
//6.0
广播流
广播流则可以允许多个监听器存在，就如同广播一样，凡是监听了广播流，每个监听器都能获取到数据。要注意，如果在触发事件时将监听者正添加到广播流，则该监听器将不会接收当前正在触发的事件。如果取消监听，监听者会立即停止接收事件。
有两种方式创建广播流，一种直接从Stream创建，另一种使用StreamController创建。
test() async{
// 调用 Stream 的 asBroadcastStream 方法创建
Stream<int> stream = Stream<int>.periodic(Duration(seconds: 1), (e)=>e)
.asBroadcastStream();
stream = stream.take(5);
stream.listen(print);
stream.listen(print);
}
test() async{
// 创建广播流
StreamController sc = StreamController.broadcast();
sc.stream.listen(print);
sc.stream.listen(print);
sc.add("event1");
sc.add("event2");
}
