良匠网站建设百度平台客服

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