国产97色在线 | 日韩,free性玩弄少妇hd性老妇

RxJS 利用類似於隨時(shí)間推移填充的數(shù)組的可觀察對象來促進(jìn)函數(shù)式反應(yīng)式編程 (FRP)，從而允許在應(yīng)用程序中進(jìn)行更聲明式和強(qiáng)大的錯誤處理。
RxJS 中簡單流的核心操作，例如 map()、filter()、reduce() 和 take()，鏡像數(shù)組操作，但應(yīng)用於隨時(shí)間發(fā)出值的數(shù)流。
像 flatMap() 和 switch() 這樣的專用函數(shù)對於分別處理複雜數(shù)據(jù)結(jié)構(gòu)和管理多個(gè)流至關(guān)重要，這使得它們對於高級反應(yīng)式編程任務(wù)至關(guān)重要。
可以使用 concat()、merge() 和 combineLatest() 等運(yùn)算符有效地組合多個(gè)流，每個(gè)運(yùn)算符在流管理和數(shù)據(jù)同步中發(fā)揮不同的作用。
takeUntil() 函數(shù)提供了一種基於外部條件取消訂閱的機(jī)制，這說明了 RxJS 在流控制和資源管理方面的靈活性。

反應(yīng)式編程

反應(yīng)式編程是一種編程範(fàn)例，它將稱為可觀察對象的數(shù)據(jù)流作為其基本的編程單元。

流——或 RxJS 行話中的可觀察對象——類似於事件監(jiān)聽器：兩者都等待某些事情發(fā)生，並在發(fā)生時(shí)通知您。從 onClick 監(jiān)聽器獲得的一系列異步通知是數(shù)據(jù)流的完美示例。

換句話說，可觀察對像只不過是一個(gè)隨時(shí)間推移填充的數(shù)組。

該數(shù)組的元素可以來自幾乎任何地方：文件系統(tǒng)、DOM 事件、API 調(diào)用，甚至轉(zhuǎn)換後的同步數(shù)據(jù)，如數(shù)組。從根本上說，反應(yīng)式編程只不過是用可觀察對像作為程序的構(gòu)建塊。

與數(shù)組的關(guān)係

數(shù)組很簡單，因?yàn)樗鼈兊膬?nèi)容是最終的，除非明確更改。從這個(gè)意義上說，數(shù)組中沒有什麼本質(zhì)上的時(shí)間性。

另一方面，可觀察對象由時(shí)間定義。您最多只能知道流到目前為止已接收 [1, 2, 3]。您不能確定您是否會得到 4——或者不會——並且是數(shù)據(jù)源，而不是您的程序，決定了這一點(diǎn)。

流和數(shù)組之間的關(guān)係是如此深刻，以至於大多數(shù)反應(yīng)式擴(kuò)展都源於函數(shù)式編程的世界，其中列表操作是核心。

熟悉 RxJS

考慮一下常見的待辦事項(xiàng)應(yīng)用程序。讓我們看看使用 RxJS 如何顯示用戶未完成任務(wù)的名稱的問題：

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

到目前為止，這只不過是數(shù)組擴(kuò)展，但它演示了反應(yīng)式編程的函數(shù)式風(fēng)格。

通過添加更複雜、“現(xiàn)實(shí)世界”的功能，其聲明性性質(zhì)變得清晰。假設(shè)我們想要：

響應(yīng)用戶選擇查看已完成或未完成的任務(wù)來啟動請求；
每秒只發(fā)送對上次選擇的請求一次，以免在用戶快速更改選擇時(shí)浪費(fèi)帶寬；
最多重試三次失敗的請求；以及
只有當(dāng)服務(wù)器發(fā)送與上次不同的響應(yīng)時(shí)才重繪視圖。

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

逐步分解：

parameter_stream 告訴我們用戶是否想要已完成或未完成的任務(wù)，並將選擇存儲在 parameter 中；
debounce() 確保我們每秒鐘只關(guān)注最後一次按鈕點(diǎn)擊；
getTasks() 周圍的部分與之前相同；
distinctUntilChanged() 確保我們只在服務(wù)器的響應(yīng)與上次不同時(shí)才關(guān)注；以及
update() 負(fù)責(zé)更新 UI 以反映我們從服務(wù)器獲得的內(nèi)容。

在命令式、基於回調(diào)的樣式中處理 debounce、retry 和“distinct until changed”邏輯是有效的，但它既脆弱又復(fù)雜。

關(guān)鍵在於，使用 RxJS 進(jìn)行編程允許：

聲明式程序；
可擴(kuò)展的系統(tǒng)；以及
簡單直接、強(qiáng)大的錯誤處理。

在瀏覽 RxJS 十大必知函數(shù)的過程中，我們將遇到上述示例中的每個(gè)函數(shù)。

簡單流操作

簡單流（發(fā)出簡單值，如字符串的流）的基本函數(shù)包括：

map()
filter()
reduce()
take() / takeWhile()

除了 take() 和 takeWhile() 之外，這些都類似於 JavaScript 的高階數(shù)組函數(shù)。

我們將通過解決一個(gè)示例問題來應(yīng)用這些函數(shù)：查找數(shù)據(jù)庫中所有具有 .com 或 .org 網(wǎng)站的用戶，併計(jì)算其網(wǎng)站名稱的平均長度。

JSONPlaceholder 將作為我們的用戶來源。這是我們將使用的用戶數(shù)據(jù)的 JSON 表示。

1. 使用 map() 轉(zhuǎn)換數(shù)據(jù)

在可觀察對像上使用 map() 與在數(shù)組上使用它相同。它：

接受回調(diào)作為參數(shù)；
在您調(diào)用的數(shù)組的每個(gè)元素上執(zhí)行它；以及
返回一個(gè)新數(shù)組，其中原始數(shù)組的每個(gè)元素都被調(diào)用回調(diào)在其上產(chǎn)生的結(jié)果所替換。

在可觀察對像上使用 map() 時(shí)，唯一的區(qū)別是：

它返回一個(gè)新的可觀察對象，而不是一個(gè)新的數(shù)組；以及
它在可觀察對象發(fā)出新項(xiàng)目時(shí)執(zhí)行，而不是立即全部執(zhí)行一次。

我們可以使用 map() 將我們的用戶數(shù)據(jù)流轉(zhuǎn)換為僅包含其網(wǎng)站名稱的列表：

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

在這裡，我們使用 map 將傳入流中的每個(gè)用戶對象替換為每個(gè)用戶的網(wǎng)站。

RxJS 還允許您調(diào)用 map() as select()。這兩個(gè)名稱都指代相同的函數(shù)。

2. 過濾結(jié)果

像 map() 一樣，filter() 在可觀察對像上的作用與在數(shù)組上的作用大致相同。要查找每個(gè)具有 .net 或 .org 網(wǎng)站地址的用戶，我們可以這樣寫：

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

這將只選擇其網(wǎng)站以“net”或“org”結(jié)尾的用戶。

filter() 也有別名 where()。

3. 使用 reduce() 收集結(jié)果

reduce() 允許我們使用所有單個(gè)值並將它們轉(zhuǎn)換為單個(gè)結(jié)果。

reduce() 往往是最令人困惑的基本列表操作，因?yàn)榕c filter() 或 map() 不同，它的行為因使用而異。

通常，reduce() 獲取值的集合，並將其轉(zhuǎn)換為單個(gè)數(shù)據(jù)點(diǎn)。在我們的例子中，我們將向它提供一個(gè)網(wǎng)站名稱流，並使用 reduce() 將該流轉(zhuǎn)換為一個(gè)對象，該對象計(jì)算我們找到的網(wǎng)站數(shù)量以及其名稱長度的總和。

source.
  map((user) => user.website)

在這裡，我們將流簡化為單個(gè)對象，它跟蹤：

我們已經(jīng)看到了多少個(gè)站點(diǎn)；以及
所有名稱的總長度。

請記住，reduce() 只有在源可觀察對象完成時(shí)才返回結(jié)果。如果您想在每次流接收新項(xiàng)目時(shí)都知道累加器的狀態(tài)，請改用 scan()。

4. 使用 take() 限制結(jié)果

take() 和 takeWhile() 補(bǔ)充了簡單流的基本函數(shù)。

take(n) 從流中讀取 n 個(gè)值，然後取消訂閱。

我們可以使用 scan() 在每次我們收到網(wǎng)站時(shí)發(fā)出我們的對象，並且只 take() 前兩個(gè)值。

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'));
})

RxJS 還提供 takeWhile()，它允許您在某個(gè)布爾測試成立之前獲取值。我們可以這樣使用 takeWhile() 來編寫上述流：

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'))).
  reduce((data, website) => {
    return {
      count       : data.count += 1,
      name_length : data.name_length += website.length
    }
  }, { count : 0, name_length : 0 })

高階流操作

除了它們在可觀察對象而不是數(shù)組上工作之外，這些函數(shù)幾乎與熟悉的列表操作相同。

“[I]f you know how to program against Arrays using the Array#extras, then you already know how to use RxJS!” ~ RxJS 文檔

正如數(shù)組可以包含比簡單值（如數(shù)組或?qū)ο螅└}雜的數(shù)據(jù)一樣，可觀察對像也可以發(fā)出高階數(shù)據(jù)，如 Promise 或其他可觀察對象。這就是更專業(yè)的工具發(fā)揮作用的地方。

5. 使用 flatMap() 壓平流

……事實(shí)上，我們已經(jīng)在使用了！

當(dāng)我們定義源流時(shí)，我們調(diào)用了 fromPromise() 和 flatMap()：

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'))).
  scan((data, website) => {
      return {
        count       : data.count += 1,
        name_length : data.name_length += website.length
      }
    }, { count : 0, name_length : 0 }).
  take(2);

這使用了三個(gè)新的機(jī)制：

fromPromise；
Rx.Observable.from；以及
flatMap。

來自 promise 的可觀察對象

Promise 代表我們將異步獲得的單個(gè)未來值——例如，對服務(wù)器的調(diào)用的結(jié)果。

Promise 的一個(gè)定義特徵是它只代表一個(gè)未來的值。它不能返回多個(gè)異步數(shù)據(jù)；這是可觀察對象所做的，也是兩者之間的一個(gè)根本區(qū)別。

這意味著，當(dāng)我們使用 Rx.Observable.fromPromise() 時(shí)，我們得到一個(gè)可觀察對象，它發(fā)出單個(gè)值——或者：

Promise 解析到的值；或
Promise 拒絕的值。

當(dāng) Promise 返回字符串或數(shù)字時(shí)，我們不需要做任何特殊的事情。但是，當(dāng)它返回?cái)?shù)組時(shí)（在我們的例子中就是這樣），我們更希望創(chuàng)建一個(gè)可觀察對象，該對象發(fā)出數(shù)組的內(nèi)容，而不是數(shù)組本身作為單個(gè)值。

6. 使用 flatMap()

此過程稱為扁平化，flatMap() 會處理它。它有很多重載，但我們只使用最簡單和最常用的重載。

使用 flatMap() 時(shí)，我們：

在發(fā)出 Promise 的單值解析或拒絕的可觀察對像上調(diào)用 flatMap()；以及
傳遞一個(gè)函數(shù)來創(chuàng)建一個(gè)新的可觀察對象。

在我們的例子中，我們傳遞 Rx.Observable.from()，它從數(shù)組的值創(chuàng)建一個(gè)序列：

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

這涵蓋了我們簡短序言中的代碼：

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

RxJS 也為 flatMap() 提供了一個(gè)別名：selectMany()。

組合多個(gè)流

通常，我們將有多個(gè)需要組合的流。組合流的方法有很多，但有一些比其他的出現(xiàn)頻率更高。

7. 使用 concat() 和 merge() 組合流

連接和合併是組合流的兩種最常見方法。

連接通過發(fā)出第一個(gè)流的值直到它完成，然後發(fā)出第二個(gè)流的值來創(chuàng)建一個(gè)新流。

合併通過發(fā)出任何活動流的值來從多個(gè)流創(chuàng)建新流

想想在 Facebook Messenger 上同時(shí)與兩個(gè)人交談。 concat() 是您從雙方收到消息，但在回復(fù)另一個(gè)人之前完成與一個(gè)人的對話的情況。 merge() 就像創(chuàng)建一個(gè)群聊並同時(shí)接收兩條消息流。

source.
  map((user) => user.website)

concat() 流將首先打印 source1 的所有值，並且只有在 source1 完成後才開始打印 source2 的值。

merge() 流將根據(jù)接收到的順序打印 source1 和 source2 的值：它不會等待第一個(gè)流完成，然後再發(fā)出第二個(gè)流的值。

8. 使用 switch()

通常，我們想監(jiān)聽發(fā)出可觀察對象的可觀察對象，但只關(guān)注來自源的最新發(fā)射。

為了進(jìn)一步擴(kuò)展 Facebook Messenger 的類比，switch() 是您……好吧，根據(jù)當(dāng)前正在發(fā)送消息的人來切換您回復(fù)的人。

為此，RxJS 提供了 switch。

用戶界面為 switch() 提供了幾個(gè)很好的用例。如果我們的應(yīng)用程序每次用戶選擇他們想要搜索的內(nèi)容時(shí)都會發(fā)出請求，我們可以假設(shè)他們只想查看最新選擇的結(jié)果。因此，我們使用 switch() 只監(jiān)聽最新選擇的結(jié)果。

順便說一下，我們應(yīng)該確保不要浪費(fèi)帶寬，而只針對用戶每秒進(jìn)行的最後一次選擇訪問服務(wù)器。我們?yōu)榇耸褂玫暮瘮?shù)稱為 debounce()

如果您想朝另一個(gè)方向前進(jìn)，並且只遵守第一次選擇，則可以使用 throttle()。它具有相同的 API，但行為相反。

9. 協(xié)調(diào)流

如果我們想允許用戶搜索具有特定 ID 的帖子或用戶怎麼辦？

為了演示，我們將創(chuàng)建另一個(gè)下拉菜單，並允許用戶選擇他們想要檢索的項(xiàng)目的 ID。

有兩種情況。當(dāng)用戶：

更改任一選擇；或
更改兩個(gè)選擇。

使用 combineLatest() 響應(yīng)任一流的更改

在第一種情況下，我們需要創(chuàng)建一個(gè)流，該流使用以下內(nèi)容啟動網(wǎng)絡(luò)請求：

用戶最近選擇的端點(diǎn)；以及
用戶最近選擇的 ID。

……並在用戶更新任一選擇時(shí)執(zhí)行此操作。

這就是 combineLatest() 的用途：

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

每當(dāng)任一流發(fā)出值時(shí)，combineLatest() 都會獲取發(fā)出的值並將其與其他流發(fā)出的最後一個(gè)項(xiàng)目配對，並將該對以數(shù)組的形式發(fā)出。

這在圖表中更容易可視化：

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()

使用 zip 只響應(yīng)兩個(gè)流的更改

要等到用戶更新其對 id 和端點(diǎn)字段的選擇後，請將 combineLatest() 替換為 zip()。

同樣，這在圖表中更容易理解：

source.
  map((user) => user.website)

與 combineLatest() 不同，zip() 會等到兩個(gè)可觀察對像都發(fā)出新內(nèi)容後才會發(fā)出其更新值的數(shù)組。

10. takeUntil

最後，takeUntil() 允許我們監(jiān)聽第一個(gè)流，直到第二個(gè)流開始發(fā)出值。

source.
  map((user) => user.website).
  filter((website) => (website.endsWith('net') || website.endsWith('org'));
})

當(dāng)您需要協(xié)調(diào)流但不需要組合它們時(shí)，這很有用。

總結(jié)

僅僅向數(shù)組添加時(shí)間維度就開啟了對程序進(jìn)行全新思考的大門。

RxJS 的內(nèi)容遠(yuǎn)不止我們在這裡看到的這些，但這已經(jīng)足夠走得很遠(yuǎn)了。

從 RxJS Lite 開始，隨時(shí)準(zhǔn)備參考文檔，並抽出時(shí)間動手實(shí)踐。在您不知不覺中，一切都會看起來像一個(gè)流……因?yàn)橐磺卸际恰?

關(guān)於 RxJS 函數(shù)的常見問題解答 (FAQ)

RxJS 與傳統(tǒng) JavaScript 的主要區(qū)別是什麼？

RxJS 是一個(gè)使用可觀察對象的反應(yīng)式編程庫，用於簡化異步或基於回調(diào)的代碼的組合。這與使用更命令式編程風(fēng)格的傳統(tǒng) JavaScript 相比。關(guān)鍵區(qū)別在於它們?nèi)绾翁幚頂?shù)據(jù)——RxJS 將數(shù)據(jù)視為流，可以使用各種運(yùn)算符對其進(jìn)行操作和轉(zhuǎn)換，而傳統(tǒng) JavaScript 則以更線性的方式處理數(shù)據(jù)。

如何在 RxJS 中創(chuàng)建可觀察對象？

在 RxJS 中，您可以使用新的 Observable() 構(gòu)造函數(shù)創(chuàng)建可觀察對象。此構(gòu)造函數(shù)將一個(gè)函數(shù)作為參數(shù)，稱為訂閱者函數(shù)，該函數(shù)在最初訂閱可觀察對象時(shí)執(zhí)行。這是一個(gè)基本示例：

const task_stream =
  // 創(chuàng)建所有數(shù)據(jù)庫中任務(wù)的流
  getTasks().
    // 只獲取此用戶的任務(wù)
    filter((task) => task.user_id == user_id).
    // 獲取未完成的任務(wù)
    filter((task) => !task.completed).
    // 只獲取任務(wù)名稱
    map((task) => task.name)

/* 任務(wù)如下所示：
   task = {
    user_id   : number,
    completed : boolean,
    name      : string
   }
 */

如何在 RxJS 中處理錯誤？

RxJS 提供了幾個(gè)處理錯誤的運(yùn)算符，例如 catchError()、retry() 和 retryWhen()。 catchError() 運(yùn)算符用於捕獲可觀察流上的錯誤並返回新的可觀察對像或拋出錯誤。 retry() 運(yùn)算符可用於在發(fā)生錯誤時(shí)重新訂閱可觀察對象。 retryWhen() 運(yùn)算符類似，但它提供了對何時(shí)重試的更多控制。

如何取消 RxJS 中可觀察對象的訂閱？

當(dāng)您訂閱可觀察對象時(shí)，您會收到一個(gè) Subscription，它有一個(gè) unsubscribe() 方法。您可以調(diào)用此方法來取消可觀察對象的執(zhí)行並清理正在使用的任何資源。這是一個(gè)示例：

const task_stream =
  parameter_stream.
    debounce(1000).
    map((parameter) => {
      getTasks().
        retry(3).
        filter((task) => task.user_id === user_id).
        filter((task) => task.completed === parameter).
        map((task)    => task.name)
    }).
    flatMap(Rx.Observable.from).
    distinctUntilChanged().
    update()