談談FRP和Observable（二）

有些讀者看了上篇文章之後第一個問題就是「這貨performance如何，吃不吃內存」。彷彿他們一下子看穿了Signal/Observable的「軟肋」：低效且內存佔用高（潛台詞是能不能跟我手寫的C代碼比）。對此，我得先瞎扯幾句我的觀點。我們看一門新技術的前景，套用當前的俗話，就是：「先問對不對，再談好不好」。軟體領域很重要的一句話是：

Simplicity matters.

從Simplicity matters這個角度看，即便用它寫出的代碼效率不高（我很懷疑這一論斷），內存開銷太大（也存疑），但四十多行的幾乎無法寫錯的直觀代碼（見上一篇文章最後的typeahead的例子），無論從何種角度看都好過近千行複雜的代碼。何況，在當今這樣一個摩爾定律被打破，軟體幾乎無法坐等硬體主頻升級（Intel多久沒升過主頻了？）而獲得效率上的紅利的時代，誰能在並發/非同步這樣的場景下表現優異，誰就坐擁了天下。對於代碼而言，上層實現抽象度越大，下層並發的潛力就越大。

還有人談到Observable看上去就是在做stream processing，而nodejs本身就自帶stream，一切IO（比如network，file system，…）都是在處理stream，二者有何不同？node的stream是和unix哲學緊密契合的概念，非常好用，很簡單，容易使用，這是它的優點；但它局限在IO，通用性不如Observable，而且提供的操作也僅僅限於pipe等最基礎的操作，雖然有 event-stream 這樣的第三方庫加入了大量的實用操作（如map，join，split，merge等），但其功能豐富程度遠不如Observable，為composition所做的努力也遠不如Observable。

Observable是在思想上全面革新的一件利器，從上一篇文章大家應該能有所體會。這種思想還帶來一個很大的好處，就是：learn once, write everywhere。我們拿Observable和設計模式來類比。設計模式的思想，你學會了以後，寫java能用，寫python能用，在讀別人的代碼時，遇到某個模式，你一下子就能大概知道作者的意圖，這是設計模式作為一種思想的好處。Observable在此之上更進一步：我幫你統一思想，還幫你統一API。當你實現一個Decorator時，java的實現和C++的實現肯定因人而異，略有不同。而Observable定義了上百個API，只要相應的語言實現了這些API，那麼，C#的代碼和javascript的代碼並沒有太多語義上的區別，僅僅是語法的差別而已。你可以很容易把C#的例子轉換成javascript的例子，你也可以在前端使用javascript處理Observable，在後端使用java處理Observable，這便是

它對開發者的好處不是不言而喻的。

另外一些讀者的擔心是Observable是不是只能應用在很小的一些場景下才能應用。今天的文章本來就計劃給出更多的例子來探討FRP和Observable的應用場景。我們先舉幾個實際的例子看看Observable如何去應對，然後再做個總結。

案例剖析

案例一：處理todo list

假設我們有這樣一個應用，從資料庫里讀入之前撰寫的todo items，顯示給用戶，並且允許用戶添加新的todo。用戶可以敲 enter 或者 click Add 按鈕輸入；如果單擊任意todo item，會更改其是否完成的狀態。此外，todo list不允許重複。

這雖然是個很簡單的例子，相信每個人都會寫（原生的不會，至少會用jquery寫吧），但要寫得直觀，簡潔，並非易事；而且，代碼會東一塊，西一塊，並不統一，還很容易在事件監聽和創建/刪除節點時產生memory leak。

如果用RxJs處理，可以這麼寫：

讓我來解釋一下核心代碼：

• render在Observable addTodo$ 產生新數據的時候重繪整個list（這裡如果使用virtual dom，會大大提高performance）

• addTodo$ 是一個 Observable，我們用 Rx.Subject 生成。這是所有todo item的唯一信息源。它由幾部分組成：

  • 首先是已有的數據。這裡我們用 get_existing_list 這樣一個函數模擬資料庫讀取。

  • 然後是按回車或者點 Add 按鈕添加的 todo item。

  • 最後是在todo item上單擊，產生的新的狀態變化的todo item

• 我們對 addTodo$ 做 scan 操作，將所有歷史數據除重並聚合起來

• 最後使用 subscribe 進行 render。

代碼在：http://jsbin.com/goxulu/edit

案例二：Lazy loading

這個例子處理scroll up/down 事件，然後按需載入數據，不算很難，不多說。

代碼在：http://jsbin.com/noguzu/edit?html,js,output

上面兩例都是UI層面的，因為我個人對animation研究不多，所以就沒有獻醜將animation也加入進來。Observable在前端一個很重頭的使用是完美地同步 event + action + animation。當一個事件發生時，我們要產生一個非同步的動作，然後再用animation提升體驗。event是非同步的，處理event會引入新的非同步的action，之後再引入非同步的animation。這幾重非同步如果僅僅發生一次，或者，animation結束前不允許發生新的event，還比較容易用promise處理；但event是一個永不停歇的流，很可能下次處理event的action結束後，新的animation開始時，之前的event的animation還沒有結束。這樣的race condition被視覺化之後，體驗非常糟糕，要想圓滿處理，得花好多功夫，在各種各樣的state之間進行腦細胞絞殺式的同步。使用Observable，可以將這個過程大大簡化，你只需要挑選合適的operator就可以了。

案例三：data collection

在伺服器端，只要你勤于思考，也能發現Observable的廣闊的用武之地。比如我要做一個服務，定期從若干台伺服器中獲取（pull）資源使用使用信息。我們希望：

• 每個tick（100ms）請求一下伺服器的資源使用情況

• 如果上個tick的結果還未返回，而下個tick來臨，則忽略下個tick，不發請求

• 如果某個tick的結果出現異常（比如網路錯誤），那麼直接忽略

• 所有收集到的信息緩存5s，或者100條記錄，然後再進一步處理（可能是發給下一個服務）

下面是一個模擬的例子：

代碼在：http://jsbin.com/yudeqo/4/edit?html,js,console

在一個真實的環境下 getMetrics() 可能是一個async http request（或者更高效的話，tcp request，假設連接已經建立）。這裡我們使用一個帶有 setTimeout 的promise來模擬。真實的世界並不美好，所以我用了 boom() 來模擬潛在的失敗。

在 getMetricObservable() 里，每個tick產生時，我使用了一個 inFlight 變數來控制一個tick是否要產生一個 getMetrics() 請求。inFlight 在調用 getMetrics() 前後被設置。

tick ---t---t---t--------nfilter ---t---t------------ndo ---t---t------------nmap -----g-------g------ndo -----g-------g------nn# sideEffectninFlight FFFTTFFTTTTTTFFFFFFFn

在Observable里 do 可以用來處理side effect（副作用）。這是AOP（Aspect Oriented Programming）思想的一種體現。我們當然可以在map裡面處理 inFlight的改變，然而這樣會讓整個代碼變得很醜陋，而且失去了很多優勢（比如並發處理）。函數式編程很重要的一個思想是

把 side effect 關在籠子里

如果side effect不可避免，那麼，把它們放在集中的地方，顯式地告訴編譯器（或者庫）這段代碼有副作用，是最好的方式。這樣，那些沒有副作用的代碼，編譯器依舊可以盡量優化。

do 在Observable里，遇到上游的Observable傳過來的內容，不做任何處理，向下游傳遞，同時，在函數體內做相應的副作用的處理。比如你要 console.log 一些中間狀態，do是最好的選擇。在這段代碼里，每個 tick 或者每次 map 返回一個值，do 都相應改變一下 inFlight 的狀態。

正常情況下，當發生錯誤，錯誤會一路bubble到 subscribe Observable的地方。在這裡，我們不希望錯誤被bubble up，所以用 Rx.Observable.onErrorResumeNext（onErrorResumeNext有沒有VB的趕腳？~）來忽略錯誤。當然，你也可以 retry()，但這裡沒有必要。

這就有了一台伺服器的Observable。多台伺服器我們只需要把他們的Observable merge() 一下，然後 bufferWithTimeOrCount，就實現了我們的需求。下面是 onErrorResumeNext 和 bufferWithTimeOrCount 的 marble圖：

在這裡：

我們順帶對原始數據做了個處理，把一個帶著 {dt:…, metrics: […]} 的 object，轉化成了一個形如 [{}, …] 的數組。這樣，當你後續的處理需要單獨處理某種metric，如CPU，可以很方便處理。這裡只是展示一下，如果在Observable里要對數據做transformation，也是非常簡單的。注意，這裡我們沒有修改 data.metrics.map 里每個數據（可以這麼做但絕對不推薦！），而是使用prototype inheritance創建了新的數據（Object.create），prototype inheritance是copy-on-write的，我們這裡沒有動 metric 原有的數據，只是添加了新的數據 dt，所以實際上沒有拷貝原有數據，效率很高。

這個例子是純 Nodejs 的例子，放在 jsbin 里，只是為了大家能很直觀地運行和觀察結果。Observable在伺服器端有很多適用的場景，任何和event流相關的事情都可以考慮用其實現。

總結

處理Async並非易事。你要很小心地設計你的代碼，考慮這些情況：

• race conditions

• 內存泄漏（比如一個event handler，bind後忘記unbind）

• 管理複雜的狀態機

• 錯誤處理

而Observable能幫你減輕這些負擔，把你的精力集中在如何描述問題的解決方案上，而非如何去管理複雜的狀態，處理要命的race condition。

在處理Observable時，我們經常遇到一個數據流分解成多個數據流，或者多個數據流合併成一個數據流，而後者往往是非同步處理讓人頭疼的事情。Observable提供了一些手段，可以參考：

• 你可以concatAll，如果多個Observable的數據是要保留先後順序的（類比priority queue）

• 也可以mergeAll，如果多個Observable的數據不需要保留順序，先進先出（類比traffic merge）

• 還可以switch，你想在多個事件中僅僅處理最後發生的那個，忽略其他

有興趣的話，可以認真讀讀 http://reactivex.io 上的文檔。這個世界上，有兩樣東西必不辜負你的期待：微軟（參與）出品的文檔，以及老乾媽。

如果您覺得這篇文章不錯，請點贊。多謝！

歡迎訂閱公眾號『程序人生』（搜索微信號 programmer_life）。每篇文章都力求原汁原味，北京時間中午12點左右，美西時間下午8點左右與您相會。

推薦閱讀：