學習RxJS：導入

時間 2019-12-01

標籤學習 rxjs 導入简体版

原文原文鏈接

引子

新手們在異步編程裏跌倒時，永遠會有這麼一個經典問題：怎麼在一次異步調用裏return一個結果啊？react

老司機說要用回調函數，而後有條件判斷的嵌套回調（回調地獄）問題來了；程序員

老司機推薦用事件，而後異步流程裏有順序依賴；編程

老司機推薦用Promise，而後有順序依賴的流程裏，竟然還想訂閱事件；設計模式

老司機建議試試協程，誰知對方想要合併兩個異步調用；api

……網絡

以上，是異步編程裏要面對的一些難題，也是ReactiveX API 所致力解決的架構

是什麼

知道有 ReactiveX 這麼一回事，源於一位巨硬鐵粉的安利演示：Reactive LINQ 加持的C#，簡潔且頗具表達力；隨後，即是萬衆矚目的 Angular 2，這貨的標配大禮包裏就有RxJS，比比皆是的 api.invocation.map(...).subscribe(fn, fn, fn) 片段，讓jQuery青年們一頭霧水。框架

落伍老是很差的，林子裏的鳥都在討論FRP時，咱們也要跟上：異步

ReactiveX是Reactive Extensions的縮寫，通常簡寫爲Rx，最初是LINQ的一個擴展，由微軟的架構師Erik Meijer領導的團隊開發，在2012年11月開源，Rx是一個編程模型，目標是提供一致的編程接口，幫助開發者更方便的處理異步數據流，Rx庫支持.NET、JavaScript和C++，Rx近幾年愈來愈流行了，如今已經支持幾乎所有的流行編程語言了，Rx的大部分語言庫由ReactiveX這個組織負責維護，比較流行的有RxJava/RxJS/Rx.NET，社區網站是 reactivex.io。async

概念

ReactiveX 的自述是「An API for asynchronous programming with observable streams」，那麼，什麼是Observable，什麼又是Stream呢？

Stream

Erik Meijer發過一篇paper: 「Your Mouse Is a Database」，大概意思是說，用戶的鼠標點擊實際上是一個無窮而實時的事件序列，能夠將其視爲一個與時間線相關的數據流，咱們能夠查詢並操做這個數據流，在它可用時（或者等待數據流可用）:

在 Rx 編程中，任何數據均可以被表達爲數據流的形式，咱們要作的是，對數據流進行訂閱、查詢、過濾、打平、歸併等各類操做。

因爲是對數據流序列進行訂閱（觀察），Rx 的編程模型實際是基於 Observer pattern 和 Iterator pattern 這兩種設計模式構建的。

Observable

在Rx中，Observable就是一個序列，它按序對訂閱方（subscriber）進行值的推送，遵循一套「Don’t call us; we’ll call you.」的基本法。

基於 Observable 的模式，和傳統的 Observer pattern有兩點本質的不一樣：

Observable 只在至少有一個訂閱者時，數據纔開始流動
在數據流結束（iterator 再也不hasNext）時，Observable會發出通知（onCompleted）

怎麼用

程序員們大都被調教成了馬基雅維利主義者：「整點有用的」。因此，仍是來看看RxJS怎麼用吧：

場景

我有一個C類局域網，想要挨個ping一下網絡內的設備，看看哪些IP在線（並不知道DHCP的客戶端列表，因此得從 xxx.xxx.xxx.2 ping到 xxx.xxx.xxx.254）

思路

很明顯，ping 是一個異步的操做，這裏大概有 254 – 2 = 252 個異步操做，難點不在於異步，而在於流程控制，在RxJS 裏，能夠很方便的把Observable源進行歸併（merge），從而讓異步數據流可控且有序

RxJS方案

依賴

"dependencies": {

"ping": "^0.1.10",

"ramda": "^0.21.0",

"rx": "^4.1.0",

}

代碼

var Rx = require('rx')

var R = require('ramda')

var pingCommand = require('ping')

var config = {

timeout: 10, // 超時爲10秒

extra: ["-i 2"], // 每次發包間隔時長

}

function promisablePing(host) {

return new Promise((resolve, reject) => {

pingCommand.sys.probe(host

, isAlive => isAlive ? resolve(host) : reject(`${host}: unreachable.`)

, config)

})

}

function ping(host) {

return Rx.Observable.create(observer => {

return promisablePing(host)

.then(host => observer.onNext(host))

.then(_ => observer.onCompleted())

.catch(err => observer.onError(err))

})

}

var tasks = R.range(2, 254).map(i => ping(`192.168.50.${i}`))

Rx.Observable

.merge(...tasks)

.subscribe(

host => console.log(`pong: ${host}`),

err => console.error(err)

)

說明

代碼足夠簡單，值得說明的是：

Rx.Observable.create會建立一個Observable對象，對它進行訂閱的觀察者即create函數中回調的入值observer，觀察者有三個方法：onNext, onCompleted, and onError:
1. onNext ：在序列推送一個新值時被調用，對應到觀察者的subscribe第一個函參
2. onCompleted：序列中已無值可用，對應到觀察者的subscribe第三個函參
3. onError：序列中發生了錯誤，對應到觀察者的subscribe第二個函參
merge合併後的操做流，是一個對IP在：192.168.50.2 – 254 範圍內的設備進行Ping的操做序列，可是Observable有一個特色，就是任什麼時候候觸發了錯誤回調（即Rx.Observable.create建立那個的觀察者，進行了onError通知，從而觸發了消費者提供給subscribe函數第二個參數）那麼整個Observable序列就此結束。好比，個人C類子網就兩臺設備在線：xxx.xxx.xxx..100 和 xxx.xxx.xxx.200，而後 xxx.xxx.xxx.2 在10秒後超時報錯，那這條Observable時間線看起來就是這樣的：
BEGIN-> .100-.200---------------------[.2 error] ->END
Ramda 是一個優秀的函數式JS庫，固然，用成了lodash也不壞