1 什么是Stream?

Stream是Dart用來(lái)處理異步的API，和同樣用來(lái)處理異步的Future不同的是，Stream可以異步的返回多個(gè)結(jié)果，而Future只能返回一個(gè)，如果你對(duì)Future有疑問(wèn)，可以參考作者的上一篇文章，Dart基礎(chǔ)——Dart異步Future與事件循環(huán)Event Loop。

2 如何創(chuàng)建Stream?

1.1使用Stream的構(gòu)造方法

Stream?periodicStream?=?Stream.periodic(Duration(seconds:?2),?(num)?{
??return?num;
});

periodic構(gòu)造方法主要有兩個(gè)參數(shù)，第一個(gè)參數(shù)類型為Duration（時(shí)間間隔），第二個(gè)參數(shù)類型為Function，F(xiàn)unction每隔一個(gè)Duration（時(shí)間間隔）會(huì)被調(diào)用一次,參數(shù)num為事件調(diào)用的次數(shù)，從0開(kāi)始依次遞增。

翻閱源碼 Stream.periodic是使用Timer.periodic加_SyncStreamController實(shí)現(xiàn)的

1.2將方法的返回值聲明為Stream

Stream?timedCounter(Duration?interval,?[int?maxCount])?async*?{
??int?i?=?0;
??while?(true)?{
???//延遲interval（時(shí)間間隔）執(zhí)行一次
????await?Future.delayed(interval);
????//返回i??i++
????yield?"stream返回${i++}";
????if?(i?==?maxCount)?break;
??}
}

看到這里你可能會(huì)有一些疑問(wèn)什么是async*和yield?

yield為一個(gè)用async?*修飾返回值為Stream的函數(shù)返回一個(gè)值，它就像return,不過(guò)他不會(huì)結(jié)束函數(shù)

Stream?asynchronousNaturalsTo(n)?async*?{
??int?k?=?0;
??while?(k?}

這里涉及到了Dart的生成器函數(shù)概念，在這里你只需要簡(jiǎn)單理解yield的作用就可以了

1.3使用StreamController

??var?_controller?=?StreamController();

??var?_count?=?1;

??createStream()?{
??//函數(shù)每隔一秒調(diào)用一次
????Timer.periodic(Duration(seconds:?1),?(t)?{
??????_controller.sink.add(_count);
??????_count++;
????});
??}

我們主要使用_controller的兩個(gè)屬性，使用_controller.Stream獲取流，使用_controller.sink.add向流中添加數(shù)據(jù)，上面的例子使用定時(shí)器，每隔一秒向流中添加數(shù)據(jù)_count。

3 Stream的常用方法

接下來(lái)介紹一下Stream的常用方法

PS:以下Stream常用方法的展示都是用下面代碼創(chuàng)建的流

Stream?periodicStream?=?Stream.periodic(Duration(seconds:?1),?(num)?{
??return?num;
});

3.1 listen

listen作為使用Stream最重要的方法，主要用于監(jiān)聽(tīng)流的數(shù)據(jù)變化，每當(dāng)流的數(shù)據(jù)變化時(shí)，listen中的方法都會(huì)被調(diào)用。

????periodicStream.listen((event)?{
??????print(event);
????});

listen方法默認(rèn)參數(shù)為Function，參數(shù)中的event為上面示例中返回的num，每當(dāng)流返回新的數(shù)據(jù)時(shí)，listen方法都會(huì)被調(diào)用。

控制臺(tái)輸出如下

0
1
2
3
4

打印流返回的數(shù)據(jù)

listen的onError參數(shù)當(dāng)流出現(xiàn)錯(cuò)誤時(shí)調(diào)用。

listen的onDone參數(shù)當(dāng)流關(guān)閉時(shí)調(diào)用。

還有一個(gè)cancelOnError屬性，默認(rèn)情況下為true，可以將其設(shè)置為false以使訂閱在發(fā)生錯(cuò)誤后也能繼續(xù)進(jìn)行。

3.2 map

Stream.periodic(Duration(seconds:?1),?(num)?{
????return?num;
??}).map((num)?=>?num?*?2)

使用map將流返回的數(shù)據(jù)進(jìn)行轉(zhuǎn)換

控制臺(tái)輸出如下

0
2
4
6

3.3 asBroadcastStream()&broadcast

通過(guò)Stream的asBroadcastStream()或StreamController的broadcast將單訂閱的流轉(zhuǎn)換為多訂閱流

什么是單訂閱流和多訂閱流？

3.3.1 單訂閱流

單訂閱流顧名思義，此流只能有一個(gè)訂閱者，也就是單訂閱流的listen方法只能被調(diào)用一次，當(dāng)?shù)诙握{(diào)用單訂閱流的listen時(shí)會(huì)報(bào)錯(cuò)，值得一提的是，當(dāng)我們創(chuàng)建流時(shí)，默認(rèn)創(chuàng)建的就是單訂閱流。

3.3.2 多訂閱流

顧名思義，此流可以有多個(gè)訂閱者，也就是多訂閱流的listen方法可以被多次調(diào)用，通過(guò)Stream的asBroadcastStream()或StreamController的broadcast將單訂閱流轉(zhuǎn)換為多訂閱流。

創(chuàng)建多訂閱流

Stream?broadcastStream?=?Stream.periodic(Duration(seconds:?5),?(num)?{
??return?num;
}).asBroadcastStream();

var?_controller?=?StreamController.broadcast()

3.3.3 單訂閱流與多訂閱流的區(qū)別

第一個(gè)區(qū)別

第一個(gè)區(qū)別就是上面提到的訂閱者數(shù)量的區(qū)別

第二個(gè)區(qū)別

我們重點(diǎn)要談?wù)撘幌聝煞N流的第二個(gè)區(qū)別

第二個(gè)區(qū)別就是單訂閱流會(huì)持有自己的數(shù)據(jù)，當(dāng)訂閱者出現(xiàn)時(shí)將自身持有的數(shù)據(jù)全部返回給訂閱者，而多訂閱流不會(huì)持有任何數(shù)據(jù)，如果多訂閱流沒(méi)有訂閱者，多訂閱流會(huì)把數(shù)據(jù)丟棄。

下面我們用兩端代碼來(lái)展示兩種流處理數(shù)據(jù)上的差別

單訂閱流代碼展示

創(chuàng)建流

??var?_controller?=?StreamController.broadcast();
??var?_count?=?1;

??createStream()?{
????Timer.periodic(Duration(seconds:?1),?(t)?{
??????_controller.sink.add(_count);
??????_count++;
????});
??}

訂閱流

createStream();
Future.delayed(Duration(seconds:?5),?()?{
??_controller.stream.listen((event)?{
????print("單訂閱流$event");
????});
});

控制臺(tái)輸出如下

可以看到，單訂閱流即使前五秒我們沒(méi)有訂閱，但單訂閱流還是在持有數(shù)據(jù)，當(dāng)訂閱者出現(xiàn)時(shí)將持有的所有數(shù)據(jù)發(fā)送給訂閱者。

多訂閱流代碼展示

創(chuàng)建流

??var?_controller?=?StreamController.broadcast();
??var?_count?=?1;

??createStream()?{
????Timer.periodic(Duration(seconds:?1),?(t)?{
??????_controller.sink.add(_count);
??????_count++;
????});
??}

訂閱流

????createStream();
????Future.delayed(Duration(seconds:?5),?()?{
??????_controller.stream.listen((event)?{
????????print("多訂閱流$event");
??????});
????});
????Future.delayed(Duration(seconds:?10),?()?{
??????_controller.stream.listen((event)?{
????????print("多訂閱流二$event");
??????});
????});

控制臺(tái)輸出

可以看到多訂閱流產(chǎn)生的前五條數(shù)據(jù)都被丟棄了，只有訂閱者出現(xiàn)后生成的數(shù)據(jù)被發(fā)送給了訂閱者。

代碼看完想必你已經(jīng)理解了單訂閱流與多訂閱流的第二種區(qū)別，我制作了兩種流程圖幫助你理解

3.4 其他方法

處理 Stream 的方法

下面這些 Stream 類中的方法可以對(duì) Stream 進(jìn)行處理并返回結(jié)果：

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

4 管理流訂閱

我們可以使用StreamSubscription對(duì)象來(lái)對(duì)流的訂閱進(jìn)行管理，listen方法的返回值就是StreamSubscription對(duì)象

??StreamSubscription?subscription?=
????Stream.periodic(Duration(seconds:?1),?(num)?{
????return?num;
??}).listen((num)?{
????print(num);
??});

4.1 暫停訂閱

subscription.pause();

4.2 恢復(fù)訂閱

subscription.resume();

4.3 取消訂閱

subscription2.cancel();

當(dāng)不需要監(jiān)聽(tīng)流時(shí)記得調(diào)用這個(gè)方法，否則會(huì)造成內(nèi)存泄漏

4.4 操作流訂閱的例子

以下示例用來(lái)展示如何操作流訂閱

創(chuàng)建流

??static?var?_controller?=?StreamController();
??var?_count?=?1;
??createStream()?{
????Timer.periodic(Duration(seconds:?1),?(t)?{
??????_controller.sink.add(_count);
??????_count++;
????});
??}

創(chuàng)建監(jiān)聽(tīng)及監(jiān)聽(tīng)管理對(duì)象

??StreamSubscription?subscription2?=?_controller.stream.listen((event)?{
????print("單訂閱流$event");
??});

操作流訂閱的方法

createStream();
Future.delayed(Duration(seconds:?3),?()?{
??print("暫停");
??subscription2.pause();
});
Future.delayed(Duration(seconds:?5),?()?{
??print("繼續(xù)");
??subscription2.resume();
});
Future.delayed(Duration(seconds:?7),?()?{
??print("取消");
??subscription2.cancel();
});

輸出如下

5 在Flutter中使用StreamBuilder組件

5.1 StreamBuilder組件介紹

StreamBuilder組件主要有兩個(gè)參數(shù)

第一個(gè)參數(shù)stream，要訂閱的流

第二個(gè)參數(shù)builder，widget構(gòu)建函數(shù)

可以使用builder函數(shù)的snapshot.connectionState屬性根據(jù)流的不同狀態(tài)返回不同的組件

每當(dāng)StreamBuilder監(jiān)聽(tīng)的stream有數(shù)據(jù)變化時(shí)，builder函數(shù)就會(huì)被調(diào)用，組件重新構(gòu)建。

5.2示例代碼

import?'package:flutter/cupertino.dart';
import?'package:flutter/material.dart';
import?'package:flutter_demo/util/util.dart';

///?Copyright?(C),?2020-2020,?flutter_demo
///?FileName:?streamBuilder_demo
///?Author:?Jack
///?Date:?2020/12/27
///?Description:
class?StreamBuilderDemo?extends?StatelessWidget?{
??//創(chuàng)建流
??Stream?_stream()?{
????Duration?interval?=?Duration(seconds:?1);
????Stream?stream?=?Stream.periodic(interval,?(num)?{
??????return?num;
????});
????stream?=?stream.take(59);
????return?stream;
??}

??@override
??Widget?build(BuildContext?context)?{
????return?Scaffold(
??????appBar:?AppBar(
????????title:?Text('Stream?Demo'),
??????),
??????body:?Center(
????????child:?StreamBuilder(
??????????stream:?_stream(),
??????????builder:?(BuildContext?context,?AsyncSnapshot?snapshot)?{
????????????if?(snapshot.connectionState?==?ConnectionState.done)?{
??????????????return?Text(
????????????????'1?Minute?Completed',
????????????????style:?TextStyle(
??????????????????fontSize:?30.0,
????????????????),
??????????????);
????????????}?else?if?(snapshot.connectionState?==?ConnectionState.waiting)?{
??????????????return?Text(
????????????????'Waiting?For?Stream',
????????????????style:?TextStyle(
??????????????????fontSize:?30.0,
????????????????),
??????????????);
????????????}
????????????return?Text(
??????????????'00:${snapshot.data.toString().padLeft(2,?'0')}',
??????????????style:?TextStyle(
????????????????fontSize:?30.0,
??????????????),
????????????);
??????????},
????????),
??????),
????);
??}
}

6 完整示例

上文所有的代碼示例都在作者的GiuHub上，https://github.com/jack0-0wu/flutter_demo，里面還包含了一些常用flutter功能的展示。