Flutter | 狀態管理拓展篇——RxDart(四)

摘要： 在前一篇文章向大家介紹了一種新的狀態管理方式——BLoC，它在分離我們的ui邏輯與業務邏輯上表現十分優秀。但是在最後我們發現了一個問題。 bloc是一個典型的觀察者模式，我們以counter bloc舉例，在A,B頁面都存在觀察者，它們監聽的是同一個廣播流，當我們p...

在前一篇文章向大家介紹了一種新的狀態管理方式——BLoC，它在分離我們的ui邏輯與業務邏輯上表現十分優秀。但是在最後我們發現了一個問題。

bloc是一個典型的觀察者模式，我們以counter bloc舉例，在A,B頁面都存在觀察者，它們監聽的是同一個廣播流，當我們pop B頁面，回到A頁面這個操作不會出現任何問題，而當我們再次進入B頁面的時候卻發現，它顯示了初始值0，而不是我們想要的value，只有等我們再次按下按鈕時，它才能重新整理獲得實際的value。

Stream很棒，但是還不夠強大

所以今天要給大家簡單介紹下ReactiveX的dart 實現—— RxDart ，它極大的擴充套件了Stream的功能,能夠讓我們在使用bloc的時候更加遊刃有餘。

在正式開始介紹前，我希望您已經閱讀並理解了stream的相關知識，後面的內容都基於此。如果您還未了解過dart:stream 的話，我建議您先閱讀這篇文章：Dart:什麼是Stream。

RxDart

ReactiveX是什麼

ReactiveX是一個強大的庫，用於通過使用 可觀察序列 來編寫 非同步 和 基於事件 的程式。它突破了 語言 和 平臺 的限制，讓我們編寫非同步程式就像在自家花園散步那樣 easy。我相信你一定會愛上它！

基本概念

在Dart:什麼是Stream這篇文章中，我用到一個模型來理解stream裡面到底發生了什麼。今天我們還是利用這個模型來看看，在rxdart中它是什麼樣的。

這個模式的關鍵思維在於觀察者的無狀態。我們平時呼叫方法的時候一定是很清楚我們什麼時候呼叫，並立刻會返回一個預想的結果。

但是在這裡，我們中間進行處理的時候，完全是處於非同步狀態的，也就是說無法立刻返回一個值。我們不知道stream什麼時候會“吐”出處理結果，所以必須要一個觀察者來守著這個出口。

當有事件/資料流出時，觀察者捕捉到了這個事件並解析處理。

• Subject實現並擴充套件了StreamController,它符合StreamController的所有規範。假如您之前使用的StreamController,那麼你可以直接替換為Subject。你可以把它想像成streamController。
• Observable實現並擴充套件了Stream。它將常用的stream和streamTransformer組合成了非常好用的api。你可以把它想像成stream。

可觀察物件——Observable

建立Observavle

你可以把stream直接包裝成Observable

var obs = Observable(Stream.fromIterable([1,2,3,4,5]));

obs.listen(print);
複製程式碼

輸出：1 2 3 4 5

通過Future建立：fromFuture

var obs = Observable.fromFuture(new Future.value("Hello"));
 
obs.listen(print); 
複製程式碼

輸出：Hello

通過Iterable建立：fromIterable

var obs = Observable.fromInterable([1,2,3,4,5]);

obs.listen(print);
複製程式碼

輸出：1 2 3 4 5

讓流的“吐”出間隔一段時間：interval

interval方法能夠讓流“吐出資料”後間隔一段時間再吐下一個資料。

var obs = Observable(Stream.fromIterable([1,2,3,4,5]))
.interval(new Duration(seconds: 1));

obs.listen(print);
複製程式碼

輸出：1 ... 2 ... 3 ... 4 ... 5

其中...代表停頓了一秒。

迭代地處理資料：map

map方法能夠讓我們迭代的處理每一個數據並返回一個新的資料

var obs = Observable(Stream.fromIterable([1,2,3,4,5]))
.map((item)=>++item);

obs.listen(print);
複製程式碼

輸出：2 3 4 5 6

擴充套件流：expand

expand方法能夠讓我們把把每個item擴充套件至多個流

var obs = Observable(Stream.fromIterable([1,2,3,4,5]))
.expand((item)=> [item,item.toDouble()]);

 obs.listen(print);
複製程式碼

輸出：1 1.0 2 2.0 3 3.0 4 4.0 5 5.0

這裡我們將每個資料擴充套件成【item，item.toDouble】你可以擴充套件成任意組的流。假如這是一個廣播Observable,並被多次收聽，那麼他可以單獨呼叫expand並擴充套件自己。

合併流：merge

merge方法能夠讓我們合併多個流,請注意輸出。

var obs = Observable.merge([
Stream.fromIterable([1,2,3]),
Stream.fromIterable([4,5,6]),
Stream.fromIterable([7,8,9]),
]);

obs.listen(print);
複製程式碼

輸出：1 4 7 2 5 8 3 6 9

順序執行多個流：concat

concat方法能夠讓我們按照順序執行一組流，當一組流執行完畢後，再開始執行下一組。

var obs = Observable.concat([
Stream.fromIterable([1,2,3]),
Stream.fromIterable([4,5,6]),
Stream.fromIterable([7,8,9]),
]);

obs.listen(print);
複製程式碼

輸出：1 2 3 4 5 6 7 8 9

檢查每一個item：every

every會檢查每個item是否符合要求，然後它將會返回一個能夠被轉化為 Observable 的 AsObservableFuture< bool>。

var obs = Observable.fromIterable([1,2,3,4,5]);

obs.every((x)=> x < 10).asObservable().listen(print);
複製程式碼

輸出結果：true

關於Observable你還需要知道這些

• Dart中 Observables 預設是 單一訂閱 。如果您嘗試兩次收聽它，則會丟擲 StateError 。你可以使用工廠方法或者 asBroadcastStream 將其轉化為多訂閱流。

var obs = Observable(Stream.fromIterable([1,2,3,4,5])).asBroadcastStream();
複製程式碼

• 很多方法的返回值並不是一個 Single 也不是一個 Observable 而是必須返回一個Dart的 Future 。幸運的是你很容易找到一些方法，把他們轉化成回 stream 。
• 出現錯誤時，Dart中的Stream不會預設關閉。但是在Rxdart中，Error會導致Observable終止，除非它被運算子攔截。
• 預設情況下Dart中Stream是非同步的，而Observables預設是同步的。
• 在處理多訂閱Observable的時候，onListen方法只有在第一次會被呼叫。且各個訂閱者之間不會互相干涉。

var obs = Observable(Stream.fromIterable([1,2,3,4,5])).asBroadcastStream();

//第一個訂閱者
obs.interval(Duration(seconds: 1)).map((item) => ++item).listen(print);
//第二個訂閱者
obs.listen(print);
複製程式碼

輸出：1 2 3 4 5 2 3 4 5 6

以上是一些比較常見的Observable的使用方法，它並不完整，我將會在以後持續的更新這篇文章，並完整介紹它的功能

增強版StreamController——Subject

普通廣播流控制器：PublishSubject

PublishSubject就是一個普通廣播版StreamController，你可以多次收聽，預設是sync是false，也就是說裡面是一個 AsyncBroadcastStreamController 非同步廣播流。

快取最新一次事件的廣播流控制器：BehaviorSubject

BehaviorSubject也是一個廣播流，但是它能記錄下最新一次的事件，並在新的收聽者收聽的時候將記錄下的事件作為第一幀傳送給收聽者。

還記得我們文章開頭的那個小問題嗎？在B頁面重新收聽的時候，獲取不到最新的事件，必須等我們重新觸發流才可以得到正確的值。

我發誓我絕對不是為了湊篇幅

ok,我們現在用BehaviorSubject替換掉我們的StreamCroller

//var _countController = StreamController.broadcast<int>();

var _subject = BehaviorSubject<int>();
複製程式碼

真的就是這麼簡單，無縫替換:laughing:

程式碼已上傳 Flutter%2FFlutter-Notebook%2Ftree%2Fmaster%2Fmecury_project%2Fexample%2Fbloc_demo" rel="nofollow,noindex">github ，讓我們來看看效果

再來看兩個例子，相信大家會對BehaviorSubject理解更深刻

例1

final subject = new BehaviorSubject<int>();

subject.add(1);
subject.add(2);
subject.add(3);

subject.stream.listen(print); // prints 3
subject.stream.listen(print); // prints 3
subject.stream.listen(print);
複製程式碼

輸出：3 3 3

由於我們在add(3)之後才開始收聽，所以將會收到最新的value。

例2

final subject = new BehaviorSubject<int>(seedValue: 1);

subject.stream.listen(print); // prints 1
subject.stream.listen(print); // prints 1
subject.stream.listen(print);
複製程式碼

輸出：1 1 1

seedValue作為初始值，在後面有收聽者的時候同樣會把它當成最後一次的value傳送給收聽者。

快取更多事件的廣播流控制器：BehaviorSubject

BehabiorSubject能夠快取更多的值，預設情況下將會快取所有值，並在新的收聽的時候將記錄下的事件作為第一幀傳送給收聽者。

final subject = ReplaySubject<int>();

subject.add(1);
subject.add(2);
subject.add(3);

subject.stream.listen(print); // prints 1
subject.stream.listen(print); // prints 1
subject.stream.listen(print);
複製程式碼

輸出：1 1 1 2 2 2 3 3 3

你還可以通過maxSize控制快取個數

final subject = ReplaySubject<int>(maxSize: 2);

subject.add(1);
subject.add(2);
subject.add(3);

subject.stream.listen(print); // prints 1
subject.stream.listen(print); // prints 1
subject.stream.listen(print);
複製程式碼

輸出：2 2 2 3 3 3

自定義你的Subject

你可以通過自定義一個新的subject繼承至Subject類來獲得更加個性化的功能。這裡就不舉栗子了。:stuck_out_tongue_closed_eyes:

Subject的釋放

當你不再收聽Subject，或者Subject不再使用時，請務必釋放它。你可以呼叫subscription的cancel()方法讓某個聽眾取消收聽，或者Subject.close()，關閉整個流。

瞭解更多

下面有一些優秀的文章能夠給您更多參考

• RxDart: Magical transformations of Streams
• 上文的拉麵翻譯版
• Reactive X官網

寫在最後

以上便是RxDart篇的全部內容，它只是介紹了部分RxDart的功能，我在之後會逐漸完善它，最終整理完整。

RxDart十分強大，它讓你在處理大量非同步事件的時候感覺非常舒適。我相信每一個開發者在瞭解過它之後一定會喜歡上這個好用的庫。

如果你在使用rxdart時候有任何好的idea，或是query，歡迎在下方評論區以及我的郵箱[email protected]留言，我會在24小時內與您聯絡！

下一篇文章將會是 flutter狀態管理總結篇 ，敬請關注。