知道有 ReactiveX 這麼一回事， 源於一位巨硬鐵粉的安利演示：Reactive LINQ 加持的C#，簡潔且頗具表達力；隨後，便是萬眾矚目的 Angular 2，這貨的標配大禮包里就有RxJS，比比皆是的 api.invocation.map(...).subscribe(fn, fn, ... ...
原文地址：http://www.moye.me/2016/05/31/learning_rxjs_part_one_preliminary/
引子
新手們在非同步編程里跌倒時，永遠會有這麼一個經典問題：怎麼在一次非同步調用里return一個結果啊？
老司機說要用回調函數，然後有條件判斷的嵌套回調（回調地獄）問題來了；
老司機推薦用事件，然後非同步流程里有順序依賴；
老司機推薦用Promise，然後有順序依賴的流程里，居然還想訂閱事件；
老司機建議試試協程，誰知對方想要合併兩個非同步調用；
……
以上，是非同步編程里要面對的一些難題，也是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。
概念
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:- onNext ：在序列推送一個新值時被調用，對應到觀察者的subscribe第一個函參
- onCompleted：序列中已無值可用，對應到觀察者的subscribe第三個函參
- 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也不壞
小結
以上，只是冰山一角，下回，想聊聊基於RxJS的Web框架：Cycle.js
更多文章請移步我的blog新地址： http://www.moye.me/
