RxJS Observable - 一個奇特的函數

前言

RxJS 的 Observable 有點難理解，其實 RxJS 相關的概念都有點難理解。畢竟 RxJS 引入了響應式編程這種新的模式，會不習慣是正常的。不過總得去理解嘛，而認識新的事物時，若是可以參照一個合適的已知事物比對着，會比較容易理解吧。對於 Observable，類比 JS 中的函數，仍是比較好的。

開始

封裝

先來看一個普通函數調用的例子：

function foo() {
  console.log('process...')
}

foo()
// 輸出：
// process...

很簡單，函數 foo() 封裝了一段邏輯（這裏只是向控制檯輸出），而後經過調用函數，函數執行內部的邏輯。

再來看 RxJS Observable 的一個例子：

var foo = Rx.Observable.create(() => {
  console.log('process...')
})

foo.subscribe()
// 輸出：
// process...

上例中，經過 Rx.Observable.create() 來建立 Observable 對象，將一樣將一段代碼邏輯封裝到 Observable 對象 foo 中，而後經過 foo.subscribe() 來執行封裝的代碼邏輯。

對於普通函數和 Observable 對象，封裝的代碼邏輯在每次調用時都會從新執行一次。從這一點來看，Observable 可以和普通函數同樣實現封裝代碼進行復用。

返回值

函數調用後能夠有返回值：

function foo() {
  console.log('process...')
  return 42
}

console.log(foo())
// 輸出：
// process...
// 42

Observable 執行後也會產生值，不過和函數直接返回的方式不一樣，要經過回調函數方式獲取：

var foo = Rx.Observable.create((observer) => {
  console.log('process...')
  observer.next(42)
})

foo.subscribe(value => console.log(value))
// 輸出：
// process...
// 42

Observable 對象內部是經過 observer.next(42) 這種方式返回值，而調用方則經過回調函數來接收返回的數據。形式上比普通函數直接返回值囉嗦一些。

從調用方的角度來看，兩個過程分別是：

• 普通函數：調用 > 執行邏輯 > 返回數據
• Observable：訂閱（subscribe） > 執行邏輯 > 返回數據

從獲取返回值方式來看，調用函數是一種直接獲取數據的模式，從函數那裏「拿」（pull）數據；而 Observable 訂閱後，是要由 Observable 經過間接調用回調函數的方式，將數據「推」（push）給調用方。

這裏 pull 和 push 的重要區別在於，push 模式下，Observable 能夠決定何時返回值，以及返回幾個值（即調用回調函數的次數）。

var foo = Rx.Observable.create((observer) => {
  console.log('process...')
  observer.next(1)
  setTimeout(() => observer.next(2), 1000)
})

console.log('before')
foo.subscribe(value => console.log(value))
console.log('after')
// 輸出：
// before
// process...
// 1
// after
// 2

上面例子中，Observable 返回了兩個值，第1個值同步返回，第2個值則是過了1秒後異步返回。

也就是說，從返回值來講，Observable 相比普通函數區別在於：

• 能夠返回多個值
• 能夠異步返回值

異常處理

函數執行可能出現異常狀況，例如：

function foo() {
  console.log('process...')
  throw new Error('BUG!')
}

咱們能夠捕獲到異常狀態進行處理：

try {
   foo()
} catch(e) {
  console.log('error: ' + e)
}

對於 Observable，也有錯誤處理的機制：

var foo = Rx.Observable.create((observer) => {
  console.log('process...')
  observer.error(new Error('BUG!'))
})

foo.subscribe(
  value => console.log(value),
  e => console.log('error: ' + e)
)

Observable 的 subscribe() 方法支持傳入額外的回調函數，用於處理異常狀況。和函數執行相似，出現錯誤以後，Observable 就再也不繼續返回數據了。

subscribe() 方法還支持另外一種形式傳入回調函數：

foo.subscribe({
  next(value) { console.log(value) },
  error(e) { console.log('error: ' + e) }
})

而這種形式下，傳入的對象和 Observable 內部執行函數中的 observer 參數在形式上就比較一致了。

停止執行

Observable 內部的邏輯能夠異步多個返回值，甚至返回無數個值：

var foo = Rx.Observable.create((observer) => {
  let i = 0
  setInterval(() => observer.next(i++), 1000)
})

foo.subscribe(i => console.log(i))
// 輸出：
// 0
// 1
// 2
// ...

上面例子中，Observable 對象每隔 1 秒會返回一個值給調用方。即便調用方再也不須要數據，仍舊會繼續經過回調函數向調用推送數據。

RxJS 提供了停止 Observable 執行的機制：

var foo = Rx.Observable.create((observer) => {
  console.log('start')
  let i = 0
  let timer = setInterval(() => observer.next(i++), 1000)
  return () => {
    clearInterval(timer)
    console.log('end')
  }
})

var subscription = foo.subscribe(i => console.log(i))
setTimeout(() => subscription.unsubscribe(), 2500)
// 輸出：
// start
// 0
// 1
// 2
// end

subscribe() 方法返回一個訂閱對象（subscription），該對象上的 unsubscribe() 方法用於取消訂閱，也就是停止 Observable 內部邏輯的執行，中止返回新的數據。

對於具體的 Observable 對象是如何停止執行，則要由 Observable 在執行後返回一個用於停止執行的函數，像上面例子中的這種方式。

Observable 執行結束後，會觸發觀察者的 complete 回調，因此能夠這樣：

foo.subscribe({
  next(value) { console.log(value) },
  complete() { console.log('completed') }
})

Observable 的觀察者共有上面三種回調：

• next：得到數據
• error：處理異常
• complete：執行結束

其中 next 能夠被屢次調用，error 和 complete 最多隻有一個被調用一次（任意一個被調用後再也不觸發其餘回調）。

數據轉換

對於函數返回值，有時候咱們要轉換後再使用，例如：

function foo() {
  return 1
}

console.log(f00() * 2)
// 輸出：
// 2

對於 Observable 返回的值，也會有相似的狀況，不過一般採用下面的方式：

var foo = Rx.Observable.create((observer) => {
  let i = 0
  setInterval(() => observer.next(i++), 1000)
})

foo.map(i => i * 2).subscribe(i => console.log(i))
// 輸出：
// 0
// 2
// 4
// ...

其實 foo.map() 返回了新的 Observable 對象，上面代碼等價於：

var foo2 = foo.map(i => i * 2)
foo2.subscribe(i => console.log(i))

Observable 對象 foo2 被訂閱時執行的內部邏輯能夠簡單視爲：

function subscribe(observer) {
  let mapFn = v => v * 2
  foo.subscribe(v => {
    observer.next(mapFn(v))
  })
}

將這種對數據的處理和數組進行比較看看：

var array = [0, 1, 2, 3, 4, 5]
array.map(i => i * 2).forEach(i => console.log(i))

是否是有點像？

除了 map() 方法，Observable 還提供了多種轉換方法，如 filter() 用於過濾數據，find() 值返回第一個知足條件的數據，reduce() 對數據進行累積處理，在執行結束後返回最終的數據。這些方法和數組方法功能是相似的，只不過是對異步返回的數據進行處理。還有一些轉換方法更增強大，例如能夠 debounceTime() 能夠在時間維度上對數據進行攔截等等。

Observable 的轉換方法，本質不過是建立了一個新的 Observable，新的 Observable 基於必定的邏輯對原 Observable 的返回值進行轉換處理，而後再推送給觀察者。

總結

Observable 就是一個奇怪的函數，它有和函數相似的東西，例如封裝了一段邏輯，每次調用時都會從新執行邏輯，執行有返回數據等；也有更特殊的特性，例如數據是推送（push）的方式返回給調用方法，返回值能夠是異步，能夠返回多個值等。

不過將 Observable 視做特殊函數，至少對於理解 Observable 上是比較有幫助的。

Observable 也被視爲 data stream（數據流），這是從 Observable 能夠返回多個值的角度來看的，而數據轉換則是基於當前數據流建立新的數據流，例如：

不過上圖看到的只是數據，而將 Observable 視爲特殊函數時，不該該忘了其內部邏輯，否則數據是怎麼產生的呢。