RxJS進階——關於流的理解和應用

RxJS是微軟公司推出的響應式編程的JavaScript庫。 對於它的學習，最開始個人理解是把它當成是 能優雅地解決異步問題的lodash。 隨着學習的深刻，發現它採用了訂閱者模式，其中也帶有純函數的思想。 直到在使用了RxJS 6以後才瞭解其少有人意識到的另外一面——流。

什麼是流？這裏咱們不用專業術語來解釋，用生活中你們熟悉的的例子來類比，好比「河流」。

河流有什麼特色？ 至少有兩個特色：

有朝向。 水往低處流，河流雖然可能會蜿蜒盤旋，可是朝向是固定的，好比我國的長江和黃河就都是由西往東流。 在RxJS中數據的流向也是固定的，就是從發送者到訂閱者。基本都以下面這種形式：

from(Promise.resolve(1)) // 流的源頭
......
.subscribe(x => console.log(x)); // 流的終點
複製代碼

有分支。

大的河流通常有幹流和支流，大大小小的支流匯入幹流。

RxJS中的數據則能夠經過操做符將數據流進行聚合或拆分。

// 流的聚合
mergeMap(from(Promise.resolve(1)), from(Promise.resolve(2)))
......
.subscribe(x => console.log(x))

// 流的拆分
const obs$ = from(Promise.resolve(1)
obs$.subscribe(x => console.log(x))
obs$.subscribe(x => {
  // do sth
})/
複製代碼

RxJS 6 相對於 RxJS 5（這裏指5.5如下的版本，由於pipe函數在RxJS 5.5中做爲新特性已被引入。） 來講不只修改了一部分操做符的名稱，同時作了一個較大的改動，引入了管道（pipe）。這個改動到底有多大？

首先是寫法上的變化。 RxJS 5的這種操做符的調用方式有沒有一種似曾相識的感受？ 是的，它看上去很像JQuery那種意大利麪條式的鏈式調用 而RxJS 6和Gulp的寫法有些像，想一想Gulp是什麼？基於流的構建工具！

// RxJS 5 僞代碼
myObservable
  .map(data => data * 2)
  .switchMap(...)
  .throttle(...))
  .subscribe(...);

// RxJS 6 僞代碼
myObservable
  .pipe(map(data => data * 2), switchMap(...), throttle(...))
  .subscribe(...);
複製代碼

這種寫法上的變化就帶來了用法上的變化，之前的固定「河流」能夠經過「管道」（pipe）來控制造成靈活的「水流」。

下面舉個例子來更加形象地闡述加入管道以後流的靈活性。

如今有一個這樣的業務場景： 點擊按鈕以後發送一個請求，讓服務端開始執行任務，而後輪詢發送請求查詢任務執行狀態，根據不一樣狀態進行不一樣操做。有3種狀態"controlling"——繼續輪詢, "stop"——中止輪詢，"finish"——中止輪詢，並進行後續操做。

不考慮判斷條件，僞代碼是下面這樣子：

// 開始任務
start$().pipe(
  switchMap(() => interval(1000)), // 開始輪詢
  switchMap(() => getStatus$()), // 查詢狀態
  )
.subscribe(x => {
  // 後續操做
})
複製代碼

這段代碼有一個問題沒有解決，根據狀態進行相應操做。 先來看看這3種狀態對應的操做。

• controlling。繼續輪詢很好處理，不進行任何操做便可。
• stop。中止輪詢的操做符有3個：take，須要固定次數，這個次數無法預先肯定。takeUntil，須要建立一個額外的subject來進行中止，應該能夠實現，不過代碼量比較大。takeWhile，只需簡單的邏輯判斷便可，比較合適。
• finish。問題來了，如過咱們在管道操做符中判斷狀態的並中止流的話，那麼訂閱者將沒法收到消息，意味着後續操做沒法執行。

解決方法就是把後續操做放到管道中。代碼以下：

// 開始任務
start$().pipe(
  switchMap(() => interval(1000)), // 開始輪詢
  switchMap(() => getStatus$()), // 查詢狀態
  filter(x => x==='stop' || x==='finish') // 'controlling'狀態下繼續輪詢，其它狀態進行對應操做
  takeWhile(x => x!=='stop') // 當爲'stop'時結束輪詢
  tap(() => {
    // 後續操做
  })
  takeWhile(() => false) // 操做完成結束輪詢
  )
.subscribe();
複製代碼

如今需求變化了，在另外一段代碼中，咱們也要經過查詢狀態並根據狀態進行，可是再也不須要開始任務和輪詢了。 那麼上面的代碼查詢和操做部分能夠利用pipe方法抽取出來。

const handle = pipe(
  switchMap(() => getStatus$()), // 查詢狀態
  filter(x => x==='stop' || x==='finish') // 'controlling'狀態下繼續輪詢，其它狀態進行對應操做
  takeWhile(x => x!=='stop') // 當爲'stop'時結束輪詢
  tap(() => {
    // 後續操做
  })
  takeWhile(() => false) // 操做完成結束輪詢
)
start$().pipe(
  switchMap(() => interval(1000)), // 開始輪詢
  handle
).subscribe();
// 直接複用查詢狀態代碼和後續操做部分代碼
other.pipe(
  handle
).subscribe()
複製代碼

總結一下。RxJS比較完整的理解應該是基於流的訂閱者模式，而流的靈活性體如今可拆分和聚合，有了pipe管道的加入，流的可複用性加強，所以更容易對代碼邏輯進行抽象。

原文連接：tech.gtxlab.com/rxjs-stream…

---

做者信息：朱德龍，人和將來高級前端工程師。