Stream
- Stream用于接收异步事件
- Stream 可以接收多个异步事件
- Stream.listen()方法返回StreamSubscription 可用于取消事件订阅，取消后，不再接收事件
基本使用
Stream.fromFutures([Future.delayed(Duration(seconds: 1),(){return "事件1";}),Future.delayed(Duration(seconds: 3),(){return "事件2";}),Future.delayed(Duration(seconds: 5),(){return "事件3";})]).listen((event) {print("${DateTime.now().millisecondsSinceEpoch} >> "+event);}).onDone(() {print("${DateTime.now().millisecondsSinceEpoch} >> onDone");});Stream.fromFutures([Future.delayed(Duration(seconds: 1),(){return "事件1";}),Future.delayed(Duration(seconds: 3),(){return "事件2";}),Future.delayed(Duration(seconds: 5),(){return "事件3";})]).listen((event) {print("${DateTime.now().millisecondsSinceEpoch} >> "+event);},onError: (error){},onDone: (){});
运行结果
1703817812594 >> 事件1
1703817814583 >> 事件2 //2s后打印
1703817816574 >> 事件3 //2s后打印
1703817816578 >> onDone
另外一种使用方式 更加灵活
var streamController = StreamController();streamController.stream.listen((event) {print(event);});streamController.add("事件1");streamController.add("事件2");streamController.add("事件3");
运行结果
事件1
事件2
事件3
由结果可以看出，Stream类似于rxjava
Stream.listen()方法返回StreamSubscription 可用于取消事件订阅，取消后，不再接收事件
未取消订阅
StreamController streamController = StreamController();StreamSubscription streamSubscription = streamController.stream.listen((event) {print(event);});streamController.add("事件1");streamController.add("事件2");Future((){sleep(Duration(seconds: 2));// streamSubscription.cancel();streamController.add("事件3");});
结果
事件1
事件2
事件3
取消订阅后
StreamController streamController = StreamController();StreamSubscription streamSubscription = streamController.stream.listen((event) {print(event);});streamController.add("事件1");streamController.add("事件2");Future((){sleep(Duration(seconds: 2));streamSubscription.cancel();streamController.add("事件3");});
结果
事件1
事件2
StreamController
构造函数参数表示stream的生命周期
/**
onListen: 监听开始开始
onPause：监听暂停
onResume:监听重启
onCancel：监听取消
sync ：同步 or 异步
*/factory StreamController({void onListen()?,void onPause()?,void onResume()?,FutureOr<void> onCancel()?,bool sync = false}) {return sync? _SyncStreamController<T>(onListen, onPause, onResume, onCancel): _AsyncStreamController<T>(onListen, onPause, onResume, onCancel);}
StreamController streamController = StreamController(onListen: (){print("onListen");},onPause: (){print("onPause");},onResume: (){print("onResume");},onCancel: (){print("onCancel");},);StreamSubscription streamSubscription = streamController.stream.listen((event) {print(event);});streamController.add("事件1");streamSubscription.pause();streamSubscription.resume();streamController.add("事件2");Future((){sleep(Duration(seconds: 2));streamSubscription.cancel();streamController.add("事件3");});
运行结果：
onListen
onPause
事件1
事件2
onResume
onCancel
Stream其他构造方法
void main() async{Stream.value(1).listen((event) {print(event);});Stream.fromIterable([2,3,4,5]).listen((event) {print(event);});//结合await使用Stream stream = Stream.fromIterable([6,7,8,9]);await for(var i in stream){print(i);}
}
运行结果：
1
2
6
3
7
4
8
5
9
Stream.periodic 每隔多长时间执行一次任务
Stream.periodic(Duration(seconds: 2),(computationCount){return "它死啦，它火啦，它死啦才火啦！${computationCount}";}).take(5) //执行5次 不设置一直执行.listen((event) {print(event);});
运行结果
它死啦，它火啦，它死啦才火啦！0
它死啦，它火啦，它死啦才火啦！1
它死啦，它火啦，它死啦才火啦！2
它死啦，它火啦，它死啦才火啦！3
它死啦，它火啦，它死啦才火啦！4
async* 标记的方法称为异步生成器,yield生成单个元素，yield*生成多个元素，最终汇集成流
void main(){createStream().listen((event) {print("${DateTime.now().millisecondsSinceEpoch} >> ${event}");});
}Stream<int> createStream() async*{for(int i=0;i<10;i++){sleep(Duration(seconds: 1)); //1s钟生成一个yield i;}
}
//结果
1703821955894 >> 0
1703821956912 >> 1
1703821957921 >> 2
1703821958929 >> 3
1703821959940 >> 4
1703821960953 >> 5
1703821961960 >> 6
1703821962971 >> 7
1703821963982 >> 8
1703821964993 >> 9
)
