Flutter完整開發實戰詳解(十一、全面深入理解Stream)
作為系列文章的第十一篇，本篇將非常全面帶你瞭解 Flutter 中最關鍵的設計之一，深入原理幫助你理解 Stream 全家桶，這也許是目前 Flutter 中最全面的 Stream 分析了。
前文：
一、Stream 由淺入深
Stream 在 Flutter 是屬於非常關鍵的概念，在 Flutter 中，狀態管理除了 InheritedWidget 之外，無論 rxdart ， Bloc 模式， flutter_redux ， fish_redux 都離不開 Stream 的封裝，而事實上 Stream 並不是 Flutter 中特有的，而是 Dart 中自帶的邏輯。
通俗來說， Stream 就是事件流或者管道，事件流相信大家並不陌生，簡單的說就是： 基於事件流驅動設計程式碼，然後監聽訂閱事件，並針對事件變換處理響應 。
而在 Flutter 中，整個 Stream 設計外部暴露的物件主要如下圖，主要包含了 StreamController 、 Sink 、 Stream 、 StreamSubscription 四個物件。
圖片要換
1、Stream 的簡單使用
如下程式碼所示， Stream 的使用並不複雜，一般我們只需要：
StreamController StreamSink Stream StreamSubscription
class DataBloc {
///定義一個Controller
StreamController<List<String>> _dataController = StreamController<List<String>>();
///獲取 StreamSink 做 add 入口
StreamSink<List<String>> get _dataSink => _dataController.sink;
///獲取 Stream 用於監聽
Stream<List<String>> get _dataStream => _dataController.stream;
///事件訂閱物件
StreamSubscription _dataSubscription;
init() {
///監聽事件
_dataSubscription = _dataStream.listen((value){
///do change
});
///改變事件
_dataSink.add(["first", "second", "three", "more"]);
}
close() {
///關閉
_dataSubscription.cancel();
_dataController.close();
}
}
在設定好監聽後，之後每次有事件變化時， listen 內的方法就會被呼叫，同時你還可以通過操作符對 Stream 進行變換處理。
如下程式碼所示，是不是一股 rx 風撲面而來？
_dataStream.where(test).map(convert).transform(streamTransformer).listen(onData);
而在 Flutter 中， 最後結合 StreamBuilder , 就可以完成 基於事件流的非同步狀態控制元件 了！
StreamBuilder<List<String>>(
stream: dataStream,
initialData: ["none"],
///這裡的 snapshot 是資料快照的意思
builder: (BuildContext context, AsyncSnapshot<List<String>> snapshot) {
///獲取到資料，為所欲為的更新 UI
var data = snapshot.data;
return Container();
});
那麼問題來了， 它們內部究竟是如果實現的呢？原理是什麼？各自的作用是什麼？都有哪些特性呢？後面我們將開始深入解析這個邏輯 。
2、Stream 四天王
從上面我們知道，在 Flutter 中使用 Stream 主要有四個物件，那麼這四個物件是如何“勾搭”在一起的？他們各自又擔任什麼責職呢？
首先如下圖，我們可以從進階版的流程圖上看出 整個 Stream 的內部工作流程。
image
Flutter中 Stream 、 StreamController 、 StreamSink 和 StreamSubscription 都是 abstract 物件，他們對外抽象出介面，而內部實現物件大部分都是 _ 開頭的如 _SyncStreamController 、 ControllerStream 等私有類，在這基礎上整個流程概括起來就是：
有一個事件源叫 Stream ，為了方便控制 Stream ，官方提供了使用 StreamController 作為管理；同時它對外提供了 StreamSink 物件作為事件輸入口，可通過 sink 屬性訪問; 又提供 stream 屬性提供 Stream 物件的監聽和變換，最後得到的 StreamSubscription 可以管理事件的訂閱。
所以我們可以總結出：
- StreamController ：如類名描述，用於整個
Stream過程的控制，提供各類介面用於建立各種事件流。 - StreamSink：一般作為事件的入口，提供如
add，addStream等。 - Stream：事件源本身，一般可用於監聽事件或者對事件進行轉換，如
listen、where。 - StreamSubscription：事件訂閱後的物件，表面上用於管理訂閱過等各類操作，如
cacenl、pause，同時在內部也是事件的中轉關鍵。
回到 Stream 的工作流程上，在上圖中我們知道， 通過 StreamSink.add 新增一個事件時， 事件最後會回撥到 listen 中的 onData 方法，這個過程是通過 zone.runUnaryGuarded 執行的，這裡 zone.runUnaryGuarded 是什麼作用後面再說，我們需要知道這個 onData 是怎麼來的？
image.png
如上圖，通過原始碼我們知道：
-
1、
Stream在listen的時候傳入了onData回撥，這個回撥會傳入到StreamSubscription中，之後通過zone.registerUnaryCallback註冊得到_onData物件( 不是前面的onData回撥哦 )。 -
2、
StreamSink在新增事件是，會執行到StreamSubscription中的_sendData方法，然後通過_zone.runUnaryGuarded(_onData, data);執行 1 中得到的_onData物件，觸發listen時傳入的回撥方法。
可以看出整個流程都是和 StreamSubscription 相關的，現在我們已經知道從 事件入口到事件出口 的整個流程時怎麼運作的，那麼這個過程是**怎麼非同步執行的呢？其中頻繁出現的 zone 是什麼？
3、執行緒
首先我們需要知道，Stream 是怎麼實現非同步的？
這就需要說到 Dart 中的非同步實現邏輯了，因為 Dart 是 單執行緒應用 ，和大多數單執行緒應用一樣，Dart 是以 訊息迴圈機制 來執行的，而這裡面主要包含兩個任務佇列，一個是 microtask 內部佇列，一個是 event 外部佇列，而 microtask 的優先順序又高於 event 。
預設的在 Dart 中，如 點選、滑動、IO、繪製事件 等事件都屬於 event 外部佇列， microtask 內部佇列主要是由 Dart 內部產生，而 Stream 中的執行非同步的模式就是 scheduleMicrotask 了。
因為 microtask 的優先順序又高於 event ，所以如果 microtask 太多就可能會對觸控、繪製等外部事件造成阻塞卡頓哦。
如下圖，就是 Stream 內部在執行非同步操作過程執行流程：
image
4、Zone
那麼 Zone 又是什麼？它是哪裡來的？
在上一篇章中說過，因為 Dart 中 Future 之類的非同步操作是無法被當前程式碼 try/cacth 的，而在 Dart 中你可以給執行物件指定一個 Zone ，類似提供一個 沙箱環境 ，而在這個沙箱內，你就可以全部可以捕獲、攔截或修改一些程式碼行為，比如所有未被處理的異常。
那麼專案中預設的 Zone 是怎麼來的？在 Flutter 中， Dart 中的 Zone 啟動是在 _runMainZoned 方法 ，如下程式碼所示 _runMainZoned 的 @pragma("vm:entry-point") 註解表示該方式是給 Engine 呼叫的，到這裡我們知道了 Zone 是怎麼來的了。
///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 的非同步操作，官方的解釋是： 在此區域中使用引數執行給定操作並捕獲同步錯誤。 類似的還有 runUnary 、 runBinaryGuarded 等，所以我們知道前面提到的 zone.runUnaryGuarded 就是 Flutter 在執行的這個 zone 裡執行已經註冊的 _onData ，並捕獲異常 。
5、非同步和同步
前面我們說了 Stream 的內部執行流程，那麼同步和非同步操作時又有什麼區別？具體實現時怎麼樣的呢？
我們以預設 Stream 流程為例子， StreamController 的工廠建立可以通過 sync 指定同步還是非同步，預設是非同步模式的。 而無論非同步還是同步，他們都是繼承了 _StreamController 物件，區別還是在於 mixins 的是哪個 _EventDispatch 實現：
-
_AsyncStreamControllerDispatch -
_SyncStreamControllerDispatch
上面這兩個 _EventDispatch 最大的不同就是在呼叫 sendData 提交事件時，是直接呼叫 StreamSubscription 的 _add 方法，還是呼叫 _addPending(new _DelayedData<T>(data)); 方法的區別。
如下圖， 非同步執行的邏輯就是上面說過的 scheduleMicrotask ， 在 _StreamImplEvents 中 scheduleMicrotask 執行後，會呼叫 _DelayedData 的 perform ，最後通過 _sendData 觸發 StreamSubscription 去回撥資料 。
image
6、廣播和非廣播。
在 Stream 中又非為廣播和非廣播模式，如果是廣播模式中， StreamControlle 的實現是由如下所示實現的，他們的基礎關係如下圖所示：
-
_SyncBroadcastStreamController -
_AsyncBroadcastStreamController
i
廣播和非廣播的區別在於呼叫 _createSubscription 時，內部對介面類 _StreamControllerLifecycle 的實現，同時它們的差異在於：
-
在
_StreamController裡判斷了如果Stream是_isInitialState的，也就是訂閱過的，就直接報錯 "Stream has already been listened to." ，只有未訂閱的才建立StreamSubscription。 -
在
_BroadcastStreamController中，_isInitialState的判斷被去掉了，取而代之的是isClosed判斷，並且在廣播中,_sendData是一個forEach執行：
_forEachListener((_BufferingStreamSubscription<T> subscription) {
subscription._add(data);
});
7、Stream 變換
Stream 是支援變換處理的，針對 Stream 我們可以經過多次變化來得到我們需要的結果。那麼這些變化是怎麼實現的呢？
如下圖所示，一般操作符變換的 Stream 實現類，都是繼承了 _ForwardingStream , 在它的內部的 _ForwardingStreamSubscription 裡，會通過上一個 Pre A Stream 的 listen 新增 _handleData 回撥，之後在回撥裡再次呼叫新的 Current B Stream 的 _handleData 。
所以事件變化的本質就是， 變換都是對 Stream 的 listen 巢狀呼叫組成的。
image
同時 Stream 還有轉換為 Future , 如 firstWhere 、 elementAt 、 reduce 等操作符方法，基本都是建立一個內部 _Future 例項，然後再 listen 的回撥用呼叫 Future 方法返回。
二、StreamBuilder
如下程式碼所示, 在 Flutter 中通過 StreamBuilder 構建 Widget ，只需提供一個 Stream 例項即可，其中 AsyncSnapshot 物件為資料快照，通過 data 快取了當前資料和狀態，那 StreamBuilder 是如何與 Stream 關聯起來的呢？
StreamBuilder<List<String>>(
stream: dataStream,
initialData: ["none"],
///這裡的 snapshot 是資料快照的意思
builder: (BuildContext context, AsyncSnapshot<List<String>> snapshot) {
///獲取到資料，為所欲為的更新 UI
var data = snapshot.data;
return Container();
});
image
如上圖所示， StreamBuilder 的呼叫邏輯主要在 _StreamBuilderBaseState 中， _StreamBuilderBaseState 在 initState 、 didUpdateWidget 中會呼叫 _subscribe 方法，從而呼叫 Stream 的 listen ，然後通過 setState 更新UI，就是這麼簡單有木有？
我們常用的 setState 中其實是呼叫了 markNeedsBuild ， markNeedsBuild 內部標記 element 為 diry ，然後在下一幀 WidgetsBinding.drawFrame 才會被繪製，這可以看出 setState 並不是立即生效的哦。
三、rxdart
其實無論從訂閱或者變換都可以看出， Dart 中的 Stream 已經自帶了類似 rx 的效果，但是為了讓 rx 的使用者們更方便的使用，ReactiveX 就封裝了 rxdart 來滿足使用者的熟悉感，如下圖所示為它們的對應關係：
image
在 rxdart 中， Observable 是一個 Stream ，而 Subject 繼承了 Observable 也是一個 Stream ，並且 Subject 實現了 StreamController 的介面，所以它也具有 Controller 的作用。
如下程式碼所示是 rxdart 的簡單使用，可以看出它遮蔽了外界需要對 StreamSubscription 和 StreamSink 等的認知，更符合 rx 歷史使用者的理解。
final subject = PublishSubject<String>();
subject.stream.listen(observerA);
subject.add("AAAA1");
subject.add("AAAA2"));
subject.stream.listen(observeB);
subject.add("BBBB1");
subject.close();
這裡我們簡單分析下，以上方程式碼為例，
-
PublishSubject內部實際建立是建立了一個廣播StreamController<T>.broadcast。 -
當我們呼叫
add或者addStream時，最終會呼叫到的還是我們建立的StreamController.add。 -
當我們呼叫
onListen時，也是將回調設定到StreamController中。 -
rxdart在做變換時，我們獲取到的Observable就是 this，也就是PublishSubject自身這個Stream，而Observable一系列的變換，也是基於建立時傳入的stream物件，比如：
@override Observable<S> asyncMap<S>(FutureOr<S> convert(T value)) => Observable<S>(_stream.asyncMap(convert));
所以我們可以看出來， rxdart 只是對 Stream 進行了概念變換，變成了我們熟悉的物件和操作符，而這也是為什麼 rxdart 可以在 StreamBuilder 中直接使用的原因。
所以，到這裡你對 Flutter 中 Stream 有全面的理解了沒？
自此，第十一篇終於結束了！(///▽///)