作為系列文章的第十一篇越锈，本篇將非常全面帶你了解 Flutter 中最關(guān)鍵的設(shè)計(jì)之一，深入原理幫助你理解 Stream 全家桶可训，這也許是目前 Flutter 中最全面的 Stream 分析了昌妹。

文章匯總地址：

Flutter 完整實(shí)戰(zhàn)實(shí)戰(zhàn)系列文章專欄

Flutter 番外的世界系列文章專欄

一、Stream 由淺入深

Stream 在 Flutter 是屬于非常關(guān)鍵的概念握截，在 Flutter 中飞崖，狀態(tài)管理除了 InheritedWidget 之外，無(wú)論 rxdart谨胞，Bloc 模式固歪，flutter_redux ，fish_redux 都離不開(kāi) Stream 的封裝胯努，而事實(shí)上 Stream 并不是 Flutter 中特有的牢裳，而是 Dart 中自帶的邏輯。

通俗來(lái)說(shuō)叶沛，Stream 就是事件流或者管道蒲讯，事件流相信大家并不陌生，簡(jiǎn)單的說(shuō)就是：基于事件流驅(qū)動(dòng)設(shè)計(jì)代碼灰署，然后監(jiān)聽(tīng)訂閱事件判帮，并針對(duì)事件變換處理響應(yīng)局嘁。

而在 Flutter 中，整個(gè) Stream 設(shè)計(jì)外部暴露的對(duì)象主要如下圖晦墙，主要包含了 StreamController 导狡、Sink 、Stream 偎痛、StreamSubscription 四個(gè)對(duì)象旱捧。

圖片要換

1、Stream 的簡(jiǎn)單使用

如下代碼所示踩麦，Stream 的使用并不復(fù)雜枚赡，一般我們只需要：

• 創(chuàng)建 StreamController ，
• 然后獲取 StreamSink 用做事件入口谓谦，
• 獲取 Stream 對(duì)象用于監(jiān)聽(tīng)贫橙，
• 并且通過(guò)監(jiān)聽(tīng)得到 StreamSubscription 管理事件訂閱，最后在不需要時(shí)關(guān)閉即可反粥，看起來(lái)是不是很簡(jiǎn)單卢肃？

class DataBloc {
  ///定義一個(gè)Controller
  StreamController<List<String>> _dataController = StreamController<List<String>>();
  ///獲取 StreamSink 做 add 入口
  StreamSink<List<String>> get _dataSink => _dataController.sink;
  ///獲取 Stream 用于監(jiān)聽(tīng)
  Stream<List<String>> get _dataStream => _dataController.stream;
  ///事件訂閱對(duì)象
  StreamSubscription _dataSubscription;

  init() {
    ///監(jiān)聽(tīng)事件
    _dataSubscription = _dataStream.listen((value){
      ///do change
    });
    ///改變事件
    _dataSink.add(["first", "second", "three", "more"]);

  }

  close() {
    ///關(guān)閉
    _dataSubscription.cancel();
    _dataController.close();
  }
}

在設(shè)置好監(jiān)聽(tīng)后，之后每次有事件變化時(shí)才顿， listen 內(nèi)的方法就會(huì)被調(diào)用莫湘，同時(shí)你還可以通過(guò)操作符對(duì) Stream 進(jìn)行變換處理。

如下代碼所示郑气，是不是一股 rx 風(fēng)撲面而來(lái)幅垮？

_dataStream.where(test).map(convert).transform(streamTransformer).listen(onData);

而在 Flutter 中， 最后結(jié)合 StreamBuilder , 就可以完成 基于事件流的異步狀態(tài)控件 了尾组！

StreamBuilder<List<String>>(
    stream: dataStream,
    initialData: ["none"],
    ///這里的 snapshot 是數(shù)據(jù)快照的意思
    builder: (BuildContext context, AsyncSnapshot<List<String>> snapshot) {
      ///獲取到數(shù)據(jù)忙芒，為所欲為的更新 UI
      var data = snapshot.data;
      return Container();
    });

那么問(wèn)題來(lái)了，它們內(nèi)部究竟是如果實(shí)現(xiàn)的呢讳侨？原理是什么呵萨？各自的作用是什么？都有哪些特性呢跨跨？后面我們將開(kāi)始深入解析這個(gè)邏輯 潮峦。

2、Stream 四天王

從上面我們知道歹叮，在 Flutter 中使用 Stream 主要有四個(gè)對(duì)象跑杭，那么這四個(gè)對(duì)象是如何“勾搭”在一起的？他們各自又擔(dān)任什么責(zé)職呢咆耿？

首先如下圖德谅，我們可以從進(jìn)階版的流程圖上看出 整個(gè) Stream 的內(nèi)部工作流程。

image

Flutter中 Stream 萨螺、StreamController 窄做、StreamSink 和 StreamSubscription 都是 abstract 對(duì)象愧驱，他們對(duì)外抽象出接口，而內(nèi)部實(shí)現(xiàn)對(duì)象大部分都是 _ 開(kāi)頭的如 _SyncStreamController 椭盏、ControllerStream 等私有類组砚，在這基礎(chǔ)上整個(gè)流程概括起來(lái)就是：

有一個(gè)事件源叫 Stream，為了方便控制 Stream 掏颊，官方提供了使用 StreamController 作為管理糟红；同時(shí)它對(duì)外提供了 StreamSink 對(duì)象作為事件輸入口，可通過(guò) sink 屬性訪問(wèn); 又提供 stream 屬性提供 Stream 對(duì)象的監(jiān)聽(tīng)和變換乌叶，最后得到的 StreamSubscription 可以管理事件的訂閱盆偿。

所以我們可以總結(jié)出：

• StreamController ：如類名描述，用于整個(gè) Stream 過(guò)程的控制准浴，提供各類接口用于創(chuàng)建各種事件流事扭。
• StreamSink：一般作為事件的入口，提供如 add 乐横， addStream 等求橄。
• Stream：事件源本身，一般可用于監(jiān)聽(tīng)事件或者對(duì)事件進(jìn)行轉(zhuǎn)換葡公，如 listen 罐农、 where 。
• StreamSubscription：事件訂閱后的對(duì)象匾南，表面上用于管理訂閱過(guò)等各類操作啃匿，如 cacenl 、pause 蛆楞，同時(shí)在內(nèi)部也是事件的中轉(zhuǎn)關(guān)鍵。

回到 Stream 的工作流程上夹厌，在上圖中我們知道豹爹， 通過(guò) StreamSink.add 添加一個(gè)事件時(shí)， 事件最后會(huì)回調(diào)到 listen 中的 onData 方法矛纹，這個(gè)過(guò)程是通過(guò) zone.runUnaryGuarded 執(zhí)行的臂聋，這里 zone.runUnaryGuarded 是什么作用后面再說(shuō)，我們需要知道這個(gè) onData 是怎么來(lái)的或南？

image.png

如上圖孩等，通過(guò)源碼我們知道：

• 1、Stream 在 listen 的時(shí)候傳入了 onData 回調(diào)采够，這個(gè)回調(diào)會(huì)傳入到 StreamSubscription 中肄方，之后通過(guò) zone.registerUnaryCallback 注冊(cè)得到 _onData 對(duì)象( 不是前面的 onData 回調(diào)哦 )。

• 2蹬癌、StreamSink 在添加事件是权她，會(huì)執(zhí)行到 StreamSubscription 中的 _sendData 方法虹茶，然后通過(guò) _zone.runUnaryGuarded(_onData, data); 執(zhí)行 1 中得到的 _onData 對(duì)象，觸發(fā) listen 時(shí)傳入的回調(diào)方法隅要。

可以看出整個(gè)流程都是和 StreamSubscription 相關(guān)的蝴罪，現(xiàn)在我們已經(jīng)知道從 事件入口到事件出口 的整個(gè)流程時(shí)怎么運(yùn)作的，那么這個(gè)過(guò)程是**怎么異步執(zhí)行的呢步清？其中頻繁出現(xiàn)的 zone 是什么要门？

3、線程

首先我們需要知道廓啊，Stream 是怎么實(shí)現(xiàn)異步的暂衡？

這就需要說(shuō)到 Dart 中的異步實(shí)現(xiàn)邏輯了，因?yàn)?Dart 是 單線程應(yīng)用 崖瞭，和大多數(shù)單線程應(yīng)用一樣狂巢，Dart 是以 消息循環(huán)機(jī)制 來(lái)運(yùn)行的，而這里面主要包含兩個(gè)任務(wù)隊(duì)列书聚，一個(gè)是 microtask 內(nèi)部隊(duì)列唧领，一個(gè)是 event 外部隊(duì)列，而 microtask 的優(yōu)先級(jí)又高于 event 雌续。

默認(rèn)的在 Dart 中斩个，如 點(diǎn)擊、滑動(dòng)驯杜、IO受啥、繪制事件 等事件都屬于 event 外部隊(duì)列，microtask 內(nèi)部隊(duì)列主要是由 Dart 內(nèi)部產(chǎn)生鸽心，而 Stream 中的執(zhí)行異步的模式就是 scheduleMicrotask 了滚局。

因?yàn)?microtask 的優(yōu)先級(jí)又高于 event ，所以如果 microtask 太多就可能會(huì)對(duì)觸摸顽频、繪制等外部事件造成阻塞卡頓哦藤肢。

如下圖，就是 Stream 內(nèi)部在執(zhí)行異步操作過(guò)程執(zhí)行流程：

image

4糯景、Zone

那么 Zone 又是什么嘁圈？它是哪里來(lái)的？

在上一篇章中說(shuō)過(guò)蟀淮，因?yàn)?Dart 中 Future 之類的異步操作是無(wú)法被當(dāng)前代碼 try/cacth 的最住，而在 Dart 中你可以給執(zhí)行對(duì)象指定一個(gè) Zone，類似提供一個(gè)沙箱環(huán)境 怠惶，而在這個(gè)沙箱內(nèi)涨缚，你就可以全部可以捕獲、攔截或修改一些代碼行為甚疟，比如所有未被處理的異常仗岖。

那么項(xiàng)目中默認(rèn)的 Zone 是怎么來(lái)的逃延？在 Flutter 中，Dart 中的 Zone 啟動(dòng)是在 _runMainZoned 方法 轧拄，如下代碼所示 _runMainZoned 的 @pragma("vm:entry-point") 注解表示該方式是給 Engine 調(diào)用的揽祥，到這里我們知道了 Zone 是怎么來(lái)的了。

///Dart 中

@pragma('vm:entry-point')
// ignore: unused_element
void _runMainZoned(Function startMainIsolateFunction, Function userMainFunction) {
  startMainIsolateFunction((){
    runZoned<Future<void>>(····);
  }, null);
}

///C++ 中
if (tonic::LogIfError(tonic::DartInvokeField(
          Dart_LookupLibrary(tonic::ToDart("dart:ui")), "_runMainZoned",
          {start_main_isolate_function, user_entrypoint_function}))) {
    FML_LOG(ERROR) << "Could not invoke the main entrypoint.";
    return false;
}

那么 zone.runUnaryGuarded 的作用是什么檩电？相較于 scheduleMicrotask 的異步操作拄丰，官方的解釋是：在此區(qū)域中使用參數(shù)執(zhí)行給定操作并捕獲同步錯(cuò)誤。 類似的還有 runUnary 俐末、 runBinaryGuarded 等料按，所以我們知道前面提到的 zone.runUnaryGuarded 就是 Flutter 在運(yùn)行的這個(gè) zone 里執(zhí)行已經(jīng)注冊(cè)的 _onData，并捕獲異常卓箫。

5载矿、異步和同步

前面我們說(shuō)了 Stream 的內(nèi)部執(zhí)行流程，那么同步和異步操作時(shí)又有什么區(qū)別烹卒？具體實(shí)現(xiàn)時(shí)怎么樣的呢闷盔？

我們以默認(rèn) Stream 流程為例子， StreamController 的工廠創(chuàng)建可以通過(guò) sync 指定同步還是異步旅急，默認(rèn)是異步模式的逢勾。 而無(wú)論異步還是同步，他們都是繼承了 _StreamController 對(duì)象藐吮，區(qū)別還是在于 mixins 的是哪個(gè) _EventDispatch 實(shí)現(xiàn)：

• _AsyncStreamControllerDispatch

• _SyncStreamControllerDispatch

上面這兩個(gè) _EventDispatch 最大的不同就是在調(diào)用 sendData 提交事件時(shí)溺拱，是直接調(diào)用 StreamSubscription 的 _add 方法，還是調(diào)用 _addPending(new _DelayedData<T>(data)); 方法的區(qū)別谣辞。

如下圖迫摔， 異步執(zhí)行的邏輯就是上面說(shuō)過(guò)的 scheduleMicrotask， 在 _StreamImplEvents 中 scheduleMicrotask 執(zhí)行后潦闲，會(huì)調(diào)用 _DelayedData 的 perform 攒菠，最后通過(guò) _sendData 觸發(fā) StreamSubscription 去回調(diào)數(shù)據(jù) 。

image

6歉闰、廣播和非廣播。

在 Stream 中又非為廣播和非廣播模式卓起，如果是廣播模式中和敬，StreamControlle 的實(shí)現(xiàn)是由如下所示實(shí)現(xiàn)的，他們的基礎(chǔ)關(guān)系如下圖所示：

• _SyncBroadcastStreamController

• _AsyncBroadcastStreamController

i

廣播和非廣播的區(qū)別在于調(diào)用 _createSubscription 時(shí)戏阅，內(nèi)部對(duì)接口類 _StreamControllerLifecycle 的實(shí)現(xiàn)昼弟，同時(shí)它們的差異在于：

• 在 _StreamController 里判斷了如果 Stream 是 _isInitialState 的，也就是訂閱過(guò)的奕筐，就直接報(bào)錯(cuò) "Stream has already been listened to." 舱痘，只有未訂閱的才創(chuàng)建 StreamSubscription 变骡。

• 在 _BroadcastStreamController 中，_isInitialState 的判斷被去掉了芭逝，取而代之的是 isClosed 判斷塌碌，并且在廣播中, _sendData 是一個(gè) forEach 執(zhí)行：

  _forEachListener((_BufferingStreamSubscription<T> subscription) {
      subscription._add(data);
    });

7、Stream 變換

Stream 是支持變換處理的旬盯，針對(duì) Stream 我們可以經(jīng)過(guò)多次變化來(lái)得到我們需要的結(jié)果台妆。那么這些變化是怎么實(shí)現(xiàn)的呢？

如下圖所示胖翰，一般操作符變換的 Stream 實(shí)現(xiàn)類接剩，都是繼承了 _ForwardingStream , 在它的內(nèi)部的_ForwardingStreamSubscription 里，會(huì)通過(guò)上一個(gè) Pre A Stream 的 listen 添加 _handleData 回調(diào)萨咳，之后在回調(diào)里再次調(diào)用新的 Current B Stream 的 _handleData 懊缺。

所以事件變化的本質(zhì)就是，變換都是對(duì) Stream 的 listen 嵌套調(diào)用組成的培他。

image

同時(shí) Stream 還有轉(zhuǎn)換為 Future , 如 firstWhere 鹃两、 elementAt 、 reduce 等操作符方法靶壮，基本都是創(chuàng)建一個(gè)內(nèi)部 _Future 實(shí)例怔毛，然后再 listen 的回調(diào)用調(diào)用 Future 方法返回。

二腾降、StreamBuilder

如下代碼所示, 在 Flutter 中通過(guò) StreamBuilder 構(gòu)建 Widget 拣度，只需提供一個(gè) Stream 實(shí)例即可，其中 AsyncSnapshot 對(duì)象為數(shù)據(jù)快照螃壤，通過(guò) data 緩存了當(dāng)前數(shù)據(jù)和狀態(tài)抗果，那 StreamBuilder 是如何與 Stream 關(guān)聯(lián)起來(lái)的呢？

StreamBuilder<List<String>>(
    stream: dataStream,
    initialData: ["none"],
    ///這里的 snapshot 是數(shù)據(jù)快照的意思
    builder: (BuildContext context, AsyncSnapshot<List<String>> snapshot) {
      ///獲取到數(shù)據(jù)奸晴，為所欲為的更新 UI
      var data = snapshot.data;
      return Container();
    });

image

如上圖所示冤馏， StreamBuilder 的調(diào)用邏輯主要在 _StreamBuilderBaseState 中，_StreamBuilderBaseState 在 initState 寄啼、didUpdateWidget 中會(huì)調(diào)用 _subscribe 方法逮光，從而調(diào)用 Stream 的 listen，然后通過(guò) setState 更新UI墩划，就是這么簡(jiǎn)單有木有涕刚？

我們常用的 setState 中其實(shí)是調(diào)用了 markNeedsBuild ，markNeedsBuild 內(nèi)部標(biāo)記 element 為 diry 乙帮，然后在下一幀 WidgetsBinding.drawFrame 才會(huì)被繪制杜漠，這可以看出 setState 并不是立即生效的哦。

三、rxdart

其實(shí)無(wú)論從訂閱或者變換都可以看出驾茴， Dart 中的 Stream 已經(jīng)自帶了類似 rx 的效果盼樟，但是為了讓 rx 的用戶們更方便的使用，ReactiveX 就封裝了 rxdart 來(lái)滿足用戶的熟悉感锈至，如下圖所示為它們的對(duì)應(yīng)關(guān)系：

image

在 rxdart 中晨缴， Observable 是一個(gè) Stream，而 Subject 繼承了 Observable 也是一個(gè) Stream裹赴，并且 Subject 實(shí)現(xiàn)了 StreamController 的接口喜庞，所以它也具有 Controller 的作用。

如下代碼所示是 rxdart 的簡(jiǎn)單使用棋返，可以看出它屏蔽了外界需要對(duì) StreamSubscription 和 StreamSink 等的認(rèn)知延都，更符合 rx 歷史用戶的理解。

final subject = PublishSubject<String>();

subject.stream.listen(observerA);
subject.add("AAAA1");
subject.add("AAAA2"));

subject.stream.listen(observeB);
subject.add("BBBB1");
subject.close();

這里我們簡(jiǎn)單分析下睛竣，以上方代碼為例晰房，

• PublishSubject 內(nèi)部實(shí)際創(chuàng)建是創(chuàng)建了一個(gè)廣播 StreamController<T>.broadcast 。

• 當(dāng)我們調(diào)用 add 或者 addStream 時(shí)射沟，最終會(huì)調(diào)用到的還是我們創(chuàng)建的 StreamController.add殊者。

• 當(dāng)我們調(diào)用 onListen 時(shí)，也是將回調(diào)設(shè)置到 StreamController 中验夯。

• rxdart 在做變換時(shí)猖吴，我們獲取到的 Observable 就是 this，也就是 PublishSubject 自身這個(gè) Stream 挥转，而 Observable 一系列的變換海蔽，也是基于創(chuàng)建時(shí)傳入的 stream 對(duì)象，比如：

  @override
  Observable<S> asyncMap<S>(FutureOr<S> convert(T value)) =>
      Observable<S>(_stream.asyncMap(convert));

所以我們可以看出來(lái)绑谣，rxdart 只是對(duì) Stream 進(jìn)行了概念變換党窜，變成了我們熟悉的對(duì)象和操作符，而這也是為什么 rxdart 可以在 StreamBuilder 中直接使用的原因借宵。

所以幌衣，到這里你對(duì) Flutter 中 Stream 有全面的理解了沒(méi)？

自此壤玫，第十一篇終于結(jié)束了豁护！(///▽///)

資源推薦

• Github ： https://github.com/CarGuo/
• 開(kāi)源 Flutter 完整項(xiàng)目：https://github.com/CarGuo/GSYGithubAppFlutter
• 開(kāi)源 Flutter 多案例學(xué)習(xí)型習(xí)項(xiàng)目: https://github.com/CarGuo/GSYFlutterDemo
• 開(kāi)源 Fluttre 實(shí)戰(zhàn)電子書(shū)項(xiàng)目：https://github.com/CarGuo/GSYFlutterBook

完整開(kāi)源項(xiàng)目推薦：

• GSYGithubApp Flutter
• GSYGithubApp React Native
• GSYGithubAppWeex

我們還會(huì)再見(jiàn)嗎？