狠狠色丁香婷婷综合尤物/久久精品综合一区二区三区/中国有色金属学报/国产日韩欧美在线观看 - 国产一区二区三区四区五区tv

前言

這個(gè)庫在angular中已經(jīng)集成了，所以使用起來有良好的代碼提示，但是在Vue中不行，一點(diǎn)提示都沒有，下面的代碼都在Vue項(xiàng)目中使用，以此分享自己在學(xué)習(xí)的體會(huì)：

一、初始RxJS

（1）安裝與導(dǎo)入

命令

npm install rxjs

按需導(dǎo)入：

import { Observer } from "rxjs";

（2）Observable的工作

說明：?Observable可以理解成被觀察者，Observer就是觀察者，連接兩者的橋梁就是Observable對(duì)象的函數(shù)subscribe，同時(shí)RxJS中的數(shù)據(jù)流就是Observable對(duì)象，它實(shí)現(xiàn)了觀察者模式和迭代器模式，這里聊聊前者。

=> 觀察者模式 <=

解決問題：?它需要解決在一個(gè)持續(xù)產(chǎn)生事件的系統(tǒng)中，如何分割功能，讓不同模塊只需要處理一部分邏輯。

解決方法：?將邏輯分為發(fā)布者和觀察者，發(fā)布者只管生產(chǎn)事件，之后將事件上注冊(cè)一個(gè)觀察者，至于事件如何被觀察者處理它不關(guān)心；同樣觀察者只管將接收到的事件處理掉，不關(guān)心它是如何產(chǎn)生的。

與RxJS的聯(lián)系：?Observable對(duì)象就是一個(gè)發(fā)布者，通過函數(shù)subscribe將其與觀察者Observer聯(lián)系起來。

import { of } from "rxjs";
// of操作符會(huì)返回一個(gè)observable對(duì)象，將傳入的內(nèi)容依次發(fā)射出來；
// 此時(shí)scoure$就是一個(gè)發(fā)布者，它產(chǎn)生的事件就是三個(gè)整數(shù)
const scoure$ = of(1, 2, 3);
// 這里console.log作為觀察者，將傳給它的內(nèi)容輸出出來，
// 不管數(shù)據(jù)是怎么產(chǎn)生的
scoure$.subscribe(console.log);

處理步驟：
產(chǎn)生事件：這是發(fā)布者的責(zé)任，也就是Observable對(duì)象的工作。?響應(yīng)事件：這是觀察者的責(zé)任，也就是由subscribe的參數(shù)決定。 發(fā)布者如何關(guān)聯(lián)觀察者：也就是何時(shí)調(diào)用subscribe。

=> 迭代器模式 <=

說明：?它提供一個(gè)通用的接口來遍歷數(shù)據(jù)集合的對(duì)象，并且讓使用者不用關(guān)心這個(gè)數(shù)據(jù)集合是如何實(shí)現(xiàn)的。從數(shù)據(jù)消費(fèi)的角度，迭代器實(shí)現(xiàn)分為拉和推兩種，簡單理解就是拉取數(shù)據(jù)和推送數(shù)據(jù)，RxJS屬于后者，它作為迭代器的使用者，并不需要主動(dòng)去從Observable 中拉數(shù)據(jù)，而是只要subscribe上Observable對(duì)象之后，然后就能夠收到消息的推送。

=> 創(chuàng)造Observable <=

執(zhí)行過程：?創(chuàng)建一個(gè)Observable，也就是創(chuàng)建一個(gè)發(fā)布者，這個(gè)發(fā)布者接收一個(gè)onSubscribe用于與觀察者產(chǎn)生聯(lián)系，當(dāng)發(fā)布者通過subscribe將其注冊(cè)給觀察者后，這個(gè)函數(shù)就會(huì)執(zhí)行，函數(shù)的參數(shù)就是觀察者對(duì)象，對(duì)這個(gè)對(duì)象的唯一要求就是需要存在next屬性，屬性的值是一個(gè)函數(shù)，用來接收傳過來的數(shù)據(jù)

// 0.用于定義發(fā)布者
import { Observable } from "rxjs";
// 4.觸發(fā)后這個(gè)函數(shù)的參數(shù)是觀察者的一個(gè)包裝，
// ? 它與觀察者并不等價(jià)
const onSubscribe = (observer) => {
 ?observer.next(1);
 ?observer.next(2);
 ?observer.next(3);
};
// 1.這里創(chuàng)建一個(gè)發(fā)布者，它存在一個(gè)onSubscribe函數(shù)與
// ? 觀察者產(chǎn)生聯(lián)系
const source$ = new Observable(onSubscribe);
// 2.創(chuàng)建一個(gè)觀察者，有一個(gè)next屬性用于接收傳過來的值
const theObserver = {
 ?next: (item) => console.log(item),
};
// 3.通過subscribe函數(shù)將發(fā)布者和觀察者聯(lián)系起來，此時(shí)發(fā)
// ? 布者中的onSubscribe函數(shù)會(huì)被觸發(fā)
source$.subscribe(theObserver);

=> 延遲的Observable <=

舉例：?如何讓上面的例子中推送每個(gè)正整數(shù)之間有一定的時(shí)間間隔？

思考：?這個(gè)邏輯放在哪個(gè)部分更合適？

解釋：?按照分工，發(fā)布者產(chǎn)生數(shù)據(jù)，觀察者處理數(shù)據(jù)，這樣一來發(fā)布者控制推送數(shù)據(jù)的節(jié)奏也很合理。

const onSubscribe = (observer) => {
 ?let number = 1;
 ?const handle = setInterval(() => {
 ? ?observer.next(number++);
 ? ?if (number > 3) {
 ? ? ?clearInterval(handle);
 ? ?}
 ?}, 1000);
};

結(jié)論：?發(fā)布者推送數(shù)據(jù)可以有時(shí)間間隔，這樣使得異步操作十分容易，因?yàn)閷?duì)于觀察者，只需要被動(dòng)接受推送數(shù)據(jù)來處理，再不用關(guān)心數(shù)據(jù)何時(shí)產(chǎn)生。

=> 永無止境的Observable <=

說明：?其實(shí)發(fā)布者發(fā)射的數(shù)據(jù)可以是無窮的，每次發(fā)布者使用next發(fā)射出一個(gè)數(shù)據(jù)，這個(gè)數(shù)據(jù)會(huì)被觀察者接收然后消化掉，所以不會(huì)存在數(shù)據(jù)堆積；如果發(fā)布者的next方法停止調(diào)用，只能表示發(fā)布者此時(shí)不會(huì)發(fā)射數(shù)據(jù)出去，但并不代表之后不會(huì)發(fā)射數(shù)據(jù)；如果需要明確發(fā)布就不會(huì)再有新數(shù)據(jù)產(chǎn)生了，還需要多個(gè)Observable完結(jié)的方式。

const onSubscribe = (observer) => {
 ?let number = 1;
 ?const handle = setInterval(() => {
 ? ?observer.next(number++);
 ?}, 1000);
};

=> Observable的完結(jié) <=

說明：?觀察者的next方法只能表示現(xiàn)在推送的數(shù)據(jù)是什么，并不能表示后面沒有更多數(shù)據(jù)了，也就是沒辦法完全停止它推送數(shù)據(jù)，但是在RxJS中，可以使用觀察者的complete方法來完成。

import { Observable } from "rxjs";
const onSubscribe = (observer) => {
 ?let number = 1;
 ?const handle = setInterval(() => {
 ? ?observer.next(number++);
 ? ?if (number > 3) {
 ? ? ?clearInterval(handle);
 ? ? ?// 使用函數(shù)完全停止數(shù)據(jù)的發(fā)送
 ? ? ?observer.complete();
 ? ?}
 ?}, 1000);
};
const source$ = new Observable(onSubscribe);
const theObserver = {
 ?next: (item) => console.log(item),
 ?// 定義函數(shù)來讓發(fā)布者完全停止數(shù)據(jù)的傳輸
 ?complete: () => console.log("No More Data"),
};
source$.subscribe(theObserver);

=> 錯(cuò)誤的Observable <=

說明：?理想情況下，發(fā)布者只管生產(chǎn)數(shù)據(jù)給觀察者來消耗，但是，難免有時(shí)候發(fā)布者會(huì)遇到了異常情況，而且這種異常情況不是生產(chǎn)者所能夠處理并恢復(fù)正常的，發(fā)布者在這時(shí)候沒法再正常工作了，就需要通知對(duì)應(yīng)的觀察者發(fā)生了這個(gè)異常情況，如果只是簡單地調(diào)用 complete，觀察者只會(huì)知道沒有更多數(shù)據(jù)，卻不知道沒有更多數(shù)據(jù)的原因是因?yàn)樵庥隽水惓＃裕覀冞€要在發(fā)布者和觀察者的交流渠道中增加一個(gè)新的函數(shù)error。

import { Observable } from "rxjs/Observable";
const onSubscribe = (observer) => {
 ?observer.next(1);
 ?// 此時(shí)發(fā)布者出現(xiàn)不能自己解決的錯(cuò)誤，調(diào)用方法通知觀察者，
 ?// 此時(shí)發(fā)布者已經(jīng)進(jìn)入完結(jié)的狀態(tài)，后面所調(diào)用的next和complete
 ?// 都會(huì)失效
 ?observer.error("Someting Wrong");
 ?observer.complete();
};
const source$ = new Observable(onSubscribe);
const theObserver = {
 ?next: (item) => console.log(item),
 ?// 用來處理錯(cuò)誤信息
 ?error: (err) => console.log(err),
 ?complete: () => console.log("No More Data"),
};
source$.subscribe(theObserver);

在RxJS中，一個(gè)發(fā)布者對(duì)象只有一種終結(jié)狀態(tài)，要么是complete，要么是error，一旦進(jìn)入出錯(cuò)狀態(tài)，這個(gè)發(fā)布者對(duì)象也就終結(jié)了，再不會(huì)調(diào)用對(duì)應(yīng)觀察者的next函數(shù)，也不會(huì)再調(diào)用觀察者的complete函數(shù)；同樣，如果一個(gè)發(fā)布者對(duì)象進(jìn)入了完結(jié)狀態(tài)，也不能再調(diào)用觀察者的next和error。
此外，一個(gè)觀察者對(duì)象，里面可以存在next、error、complete三個(gè)方法，用于接受發(fā)布者的三種不同事件，如果不關(guān)心某種事件，可以不實(shí)現(xiàn)對(duì)應(yīng)的方法；比如對(duì)于一個(gè)永遠(yuǎn)不會(huì)結(jié)束的發(fā)布者， 真的沒有必要提供complete方法，因?yàn)樗肋h(yuǎn)不會(huì)被調(diào)用到；但是對(duì)于錯(cuò)誤，觀察者是無法察覺發(fā)布者會(huì)出現(xiàn)什么錯(cuò)情況的，所以error方法還是需要。

（3）退訂Observable

說明：?有時(shí)候需要斷開發(fā)布者與觀察者之間的聯(lián)系，這個(gè)操作就叫做退訂，在發(fā)布者的onSubscribe函數(shù)執(zhí)行的時(shí)候，它可以返回一個(gè)對(duì)象，對(duì)象上可以有一個(gè)unsubscribe函數(shù)，執(zhí)行這個(gè)函數(shù)來進(jìn)行退訂操作。

import { Observable } from "rxjs";
const onSubscribe = (observer) => {
 ?let number = 1;
 ?const handle = setInterval(() => {
 ? ?observer.next(number++);
 ?}, 1000);
 ?return {
 ? ?unsubscribe: () => {
 ? ? ?clearInterval(handle);
 ? ?},
 ?};
};
const source$ = new Observable(onSubscribe);
const subscription = source$.subscribe((item) => console.log(item));
setTimeout(() => {
 ?subscription.unsubscribe();
}, 3500);

注意：?退訂函數(shù)執(zhí)行后，表示觀察者不再接受發(fā)布者推送的數(shù)據(jù)，但是發(fā)布者并沒有停止推送數(shù)據(jù)，因?yàn)榘l(fā)布者并沒有到達(dá)終結(jié)狀態(tài)，也就是沒有調(diào)用complete或者是error方法，此時(shí)只是發(fā)布者推送的數(shù)據(jù)觀察者不接收而已，看下面的例子：

import { Observable } from "rxjs";
const onSubscribe = (observer) => {
 ?let number = 1;
 ?const handle = setInterval(() => {
 ? ?// 將發(fā)布者發(fā)射的數(shù)據(jù)打印出來
 ? ?console.log("in onSubscribe ", number);
 ? ?observer.next(number++);
 ?}, 1000);
 ?return {
 ? ?unsubscribe: () => {
 ? ? ?// 這里不清除定時(shí)器，讓發(fā)布者繼續(xù)產(chǎn)生數(shù)據(jù)
 ? ? ?// clearInterval(handle);
 ? ?},
 ?};
};
const source$ = new Observable(onSubscribe);
// 每次觀察者執(zhí)行的時(shí)候打印出收到的數(shù)據(jù)
const subscription = source$.subscribe((item) => console.log(item));
setTimeout(() => {
 ?subscription.unsubscribe();
}, 3500);

發(fā)布者產(chǎn)生的事件，只有觀察者通過訂閱之后才會(huì)收到，在退訂之后就不會(huì)收到。

（4）了解兩種Observable

說明：?這里介紹的是Hot Observable和Cold Observable。

場景：?假設(shè)每個(gè)發(fā)布者對(duì)象有兩個(gè)觀察者對(duì)象來訂閱， 而且這兩個(gè)觀察者對(duì)象并不是同時(shí)訂閱，第一個(gè)觀察者對(duì)象訂閱N秒鐘之后，第二個(gè)觀察者對(duì)象才訂閱同一個(gè)發(fā)布者對(duì)象，而且，在這N秒鐘之內(nèi)，發(fā)布者對(duì)象已經(jīng)吐出了一些數(shù)據(jù)，此時(shí)對(duì)這吐出的數(shù)據(jù)有兩種處理：

Hot Observable：只需要接受從訂閱那一刻開始發(fā)布者產(chǎn)生的數(shù)據(jù)就行；有點(diǎn)類似在電視上面看節(jié)目，你所看到的內(nèi)容是節(jié)目當(dāng)前這一刻開始的，之前的節(jié)目你是看不見的，假如你的家人跟你一起看，那么你們看到的節(jié)目是一樣的，這就可以理解為獲取數(shù)據(jù)的數(shù)據(jù)源是相同的 Cold Observable：不能錯(cuò)過，需要獲取發(fā)布者之前產(chǎn)生的數(shù)據(jù)，也就是每次都需要獲取發(fā)布者完整的數(shù)據(jù)，可以理解為每次得到的數(shù)據(jù)與之前的數(shù)據(jù)之間并不存在聯(lián)系，是相互獨(dú)立的，也就是每次會(huì)得到獨(dú)立的數(shù)據(jù)源，就像你在手機(jī)應(yīng)用市場下載游戲，跟你在同樣地方下載的游戲是一樣的。

理解：?那么就可以得到這樣的結(jié)果，如果Cold Observable沒有訂閱者，數(shù)據(jù)不會(huì)真正的產(chǎn)生，就像你如果不主動(dòng)下載游戲，你手機(jī)上不可能玩到的；而對(duì)于Hot Observable在沒有訂閱者的時(shí)候，數(shù)據(jù)依然產(chǎn)生，只不過不傳入數(shù)據(jù)管道而已，就像電視機(jī)節(jié)目，節(jié)目一直存在與此，只是你沒切換到那個(gè)頻道觀看而已。

（5）操作符簡介

說明：?一個(gè)發(fā)布者對(duì)象就是一個(gè)數(shù)據(jù)流，在RxJS中數(shù)據(jù)流一般使用$開頭來命名；在一個(gè)復(fù)雜問題里面，數(shù)據(jù)流并不會(huì)直接交給觀察者來處理，在這途中會(huì)使用一系列內(nèi)置的函數(shù)來處理數(shù)據(jù)，這些函數(shù)可以理解為操作符；就像一個(gè)管道，數(shù)據(jù)從管道的一段流入，途徑管道各個(gè)環(huán)節(jié)，當(dāng)數(shù)據(jù)到達(dá)觀察者的時(shí)候，已經(jīng)被管道操作過，有的數(shù)據(jù)已經(jīng)被中途過濾拋棄掉了，有的數(shù)據(jù)已經(jīng)被改變了原來的形態(tài)，而且最后的數(shù)據(jù)可能來自多個(gè)數(shù)據(jù)源，最后觀察者只需要處理能夠達(dá)到終點(diǎn)的數(shù)據(jù)。

說明：?在數(shù)據(jù)管道中流淌的數(shù)據(jù)就像是水，從上游流向下游。對(duì)一個(gè)操作符來說，上游可能是一個(gè)數(shù)據(jù)源，也可能是其他操作符，下游可能是最終的觀察者，也可能是另一個(gè)操作符，每一個(gè)操作符之間都是獨(dú)立的，正因?yàn)槿绱耍钥梢詫?duì)操作符進(jìn)行任意組合，從而產(chǎn)生各種功能的數(shù)據(jù)管道。

6）理解彈珠圖

作用：?RxJS中每一個(gè)發(fā)布者是一個(gè)數(shù)據(jù)流，簡單的數(shù)據(jù)流可以由大腦想象出來，但是復(fù)雜的可就不好像了，此時(shí)就可以使用彈珠圖來具體的方式來描述數(shù)據(jù)流，看兩張圖：

說明：?這個(gè)彈珠圖所表示的數(shù)據(jù)流，每間隔一段時(shí)間吐出一個(gè)遞增的正整數(shù)，吐出到3的時(shí)候結(jié)束。因?yàn)槊恳粋€(gè)吐出來的數(shù)據(jù)都像是一個(gè)彈珠，所以這種表達(dá)方式叫做彈珠圖。在彈珠圖中，每個(gè)彈珠之間的間隔，代表的是吐出數(shù)據(jù)之間的時(shí)間間隔，通過這種形式，能夠很形象地看清楚每個(gè)發(fā)布者對(duì)象中數(shù)據(jù)的分布。 根據(jù)彈珠圖的傳統(tǒng)，豎杠符號(hào)|代表的是數(shù)據(jù)流的完結(jié)，對(duì)應(yīng)調(diào)用complete函數(shù)，數(shù)據(jù)流吐出數(shù)據(jù)3之后立刻就完結(jié)了。 符號(hào)×代表數(shù)據(jù)流中的異常，對(duì)應(yīng)于調(diào)用下游的error函數(shù)。

注意：?為了描述操作符的功能，彈珠圖中往往會(huì)出現(xiàn)多條時(shí)間軸，因?yàn)楦鞑糠植僮鞣墓ぷ鞫际前焉嫌蔚臄?shù)據(jù)轉(zhuǎn)為傳給下游的數(shù)據(jù)，在彈珠圖上必須把上下游的數(shù)據(jù)流都展現(xiàn)出來，此外，編寫彈珠圖可以去此處，后面如果存在彈珠圖的地方所使用的代碼復(fù)制到此處就可以看到了。

二、實(shí)現(xiàn)操作符

理解：?一個(gè)操作符是返回一個(gè)Observable對(duì)象的函數(shù)，不過，有的操作符是根據(jù)其他Observable對(duì)象產(chǎn)生返回的Observable對(duì)象，有的操作符則是利用其他類型輸出產(chǎn)生返回的Observable對(duì)象，還有一些操作符不需要輸出就可以憑空創(chuàng)造一個(gè)Observable對(duì)象，這里以實(shí)現(xiàn)一個(gè)操作符來慢慢理解什么是操作符。

（1）實(shí)現(xiàn)操作符函數(shù)

說明：?每一個(gè)操作符是一個(gè)函數(shù)，不管函數(shù)的功能是怎樣的，它需要包含以下功能點(diǎn)，這里實(shí)現(xiàn)map操作符為例

返回?個(gè)全新的Observable對(duì)象。 需要存在訂閱和退訂的操作。 處理異常情況。 及時(shí)釋放資源。

=> 返回Observable對(duì)象 <=

分析：?首先map操作符的功能是遍歷得到的數(shù)據(jù)，通過傳入的參數(shù)函數(shù)來處理這些數(shù)據(jù)，看下面的例子：

// 這里的函數(shù)參數(shù)將數(shù)據(jù)的每一個(gè)值都乘以2，
// 如果source$是?個(gè) 1、2、3的序列，
// 那么map返回的序列就是2、4、6，根據(jù)函數(shù)式編程的原則，
// map函數(shù)是不會(huì)修改原始的數(shù)據(jù)的，同時(shí)其返回值是?個(gè)全
// 新的Observable對(duì)象，這樣可以保持原始Observable對(duì)象的狀態(tài)
// 避免不可預(yù)料的行為
const result$ = source$.map(x => x * 2);

實(shí)現(xiàn)：?根據(jù)上面的分析可以得到下面這個(gè)函數(shù)

// 這里的project就是傳遞給map操作符的函數(shù)參數(shù)
function map(project) {
 ?// map中利?new關(guān)鍵字創(chuàng)造了?個(gè)Observable對(duì)象，
 ?// 函數(shù)返回的結(jié)果就是這個(gè)對(duì)象，如此?來，
 ?// map可以?于鏈?zhǔn)秸{(diào)?，可以在后?調(diào)?其他的操作符，
 ?// 或者調(diào)?subscribe增加Observer。
 ?return new Observable((observer) => {
 ? ?// 假設(shè)此處this表示發(fā)布者對(duì)象，訂閱后數(shù)據(jù)就會(huì)交給觀察者了
 ? ?this.subscribe({
 ? ? ?next: (value) => observer.next(project(value)),
 ? ? ?error: (err) => observer.error(error),
 ? ? ?complete: () => observer.complete(),
 ? ?});
 ?});
}

=> 退訂處理 <=

說明：?作為一個(gè)通用的操作符，無法預(yù)料上游Observable是如何實(shí)現(xiàn)的，上游完全可能在被訂閱時(shí)分配了特殊資源，如果不明確地告訴上游這些資源再也用不著了的話，它也不會(huì)釋放這些資源，此時(shí)就會(huì)造成資源的泄露，所以下游退訂那些資源，就要告訴上游退訂那些資源。

function map(project) {
 ?return new Observable((observer) => {
 ? ?const sub = this.subscribe({
 ? ? ?next: (value) => observer.next(project(value)),
 ? ? ?error: (err) => observer.error(error),
 ? ? ?complete: () => observer.complete(),
 ? ?});
 ? ?return {
 ? ? ?// 根據(jù)前面的了解這里需要存在一個(gè)unsubscribe
 ? ? ?// 方法用于退訂
 ? ? ?unsubscribe: () => {
 ? ? ? ?sub.unsubscribe();
 ? ? ?},
 ? ?};
 ?});
}

=> 處理異常 <=

說明：?上面代碼中的參數(shù)project可以輸入的情況有很多，可能存在執(zhí)行的時(shí)候不合理的代碼，此時(shí)就會(huì)出現(xiàn)異常，此時(shí)需要捕獲異常錯(cuò)誤，把異常錯(cuò)誤沿著數(shù)據(jù)流往下游傳遞，最終如何處理交給觀察者來決定。

function map(project) {
 ?return new Observable((observer) => {
 ? ?const sub = this.subscribe({
 ? ? ?next: (value) => {
 ? ? ? ?try {
 ? ? ? ? ?observer.next(project(value));
 ? ? ? ?} catch (err) {
 ? ? ? ? ?observer.error(err);
 ? ? ? ?}
 ? ? ?},
 ? ? ?error: (err) => observer.error(error),
 ? ? ?complete: () => observer.complete(),
 ? ?});
 ? ?return {
 ? ? ?unsubscribe: () => {
 ? ? ? ?sub.unsubscribe();
 ? ? ?},
 ? ?};
 ?});
}

（2）關(guān)聯(lián)Observable

使用原型：?這個(gè)操作符在使用的時(shí)候需要一個(gè)Observable對(duì)象實(shí)例，因此這個(gè)操作符是一個(gè)實(shí)例操作符，此時(shí)使用打補(bǔ)丁的方式關(guān)聯(lián)發(fā)布者對(duì)象的格式為Observable.prototype.操作符 = 操作符函數(shù)，既然有實(shí)例操作符，當(dāng)然也有靜態(tài)操作符，它不需要Observable實(shí)例就可以使用，它的打補(bǔ)丁的格式為Observable.操作符 = 操作符函數(shù)，這個(gè)的弊端在于會(huì)影響每一個(gè)Observable對(duì)象。

Observable.prototype.map = map;

使用call和bind：?解決上面的問題，可以讓我們?定義的操作符只對(duì)指定的 Observable對(duì)象可?，這時(shí)就可以?bind，當(dāng)然也可以使用call。

// 一般使用
const result$ = map.bind(Observable對(duì)象)(x => x * 2);
// 鏈?zhǔn)秸{(diào)用
const result$ = map.bind(
 ? ? ? ? ? ? ? ? ? ?map.bind(Observable對(duì)象)((x) => x * 2)
 ? ? ? ? ? ? ? ?)((x) => x + 1);
// 一般使用
onst result$ = map.call(Observable對(duì)象, x => x * 2);
// 鏈?zhǔn)秸{(diào)用
const result$ = map.call(
 ? ?map.call(Observable對(duì)象, (x) => x * 2),
 ? ?(x) => x * 2
);

3）改進(jìn)操作符

說明：?如果遵循函數(shù)式編程思想，需要使用純函數(shù)，也就是函數(shù)執(zhí)行的結(jié)果完全由輸入的參數(shù)決定，但是上面定義的函數(shù)中存在this，這是一個(gè)不確定的因素，也就是這個(gè)函數(shù)不屬于純函數(shù)了，所以在此處需要改進(jìn)一下。

=> 缺陷 <=

說明：?在現(xiàn)代網(wǎng)頁開發(fā)的過程中，都會(huì)經(jīng)過打包才發(fā)布到產(chǎn)品環(huán)境，在打包的過程中會(huì)使用Tree-Shaking這個(gè)工具來去除代碼中沒有使用的代碼，比如那些引入的變量但是并沒有使用這種的；但是這個(gè)工具對(duì)于RxJS來說沒什么用，這是因?yàn)門ree Shaking只能做靜態(tài)代碼檢查，并不是在程序運(yùn)行時(shí)去檢測這個(gè)函數(shù)是否被真的調(diào)用，只有這個(gè)函數(shù)在任何代碼中間都沒有引用過，才認(rèn)為這個(gè)函數(shù)不會(huì)被引用。然而，RxJS中任何一個(gè)操作符都是掛在 Observable類上或者Observable.prototype上的，賦值給Observable或者 Observable.prototype上某個(gè)屬性在Tree Shaking看來就是被引用，所以，所有的操作符，不管真實(shí)運(yùn)用時(shí)是否被調(diào)用，都會(huì)被Tree Shaking認(rèn)為是會(huì)用到的代碼，也就不會(huì)當(dāng)做死代碼刪除；其次上面關(guān)聯(lián)Observable的方式是直接添加到其原型上面，由于全局存在一個(gè)Observable對(duì)象，就跟window對(duì)象一樣，像上面添加屬性和方法是不可取的，是會(huì)帶來隱患的。

=> 不"打補(bǔ)丁" <=

說明：?開發(fā)RxJS庫的規(guī)則的其中一條就是不能使用打補(bǔ)丁的方式使操作符函數(shù)與Observable對(duì)象關(guān)聯(lián)起來。如果是實(shí)例操作符，可以使用前面介紹過的bind/call，讓一個(gè)操作符函數(shù)只對(duì)一個(gè)具體的Observable對(duì)象生效；如果是靜態(tài)操作符，直接使用就好。

// 這里使用上面實(shí)現(xiàn)的map函數(shù)實(shí)現(xiàn)一個(gè)double操作符
import { Observable, of } from "rxjs";
function map(project) {
 ?return new Observable((observer) => {
 ? ?const sub = this.subscribe({
 ? ? ?next: (value) => {
 ? ? ? ?try {
 ? ? ? ? ?observer.next(project(value));
 ? ? ? ?} catch (err) {
 ? ? ? ? ?observer.error(err);
 ? ? ? ?}
 ? ? ?},
 ? ? ?error: (err) => observer.error(error),
 ? ? ?complete: () => observer.complete(),
 ? ?});
 ? ?return {
 ? ? ?unsubscribe: () => {
 ? ? ? ?sub.unsubscribe();
 ? ? ?},
 ? ?};
 ?});
}
Observable.prototype.double = function () {
 ? ?// 將當(dāng)前的Observable對(duì)象作為this值傳遞給map函數(shù)
 ? ?return map.call(this, (x) => x * 2);
};
// of操作符用于創(chuàng)建一個(gè)Observable對(duì)象
const source$ = of(1, 2, 3);
const result$ = source$.double().subscribe((res) => console.log(res));

（4）lettable/pipeable操作符

原因：?上面使用call/bind方法在函數(shù)體內(nèi)還是會(huì)使用this，函數(shù)還是不純，其次call的返回類型是無法確定的，在ts中只能使用any表示，因此會(huì)讓其失去類型檢查。

說明：?從RxJS v5.5.0開始，加上了這種更先進(jìn)的操作符定義和使用方式，稱為pipeable操作符，也曾經(jīng)被稱為lettable操作符，但是因?yàn)樽置嫔咸y理解，所以改成pipeable。

=> let操作符 <=

作用：?實(shí)際上就是把上游的Observable對(duì)象作為參數(shù)傳遞給let操作符里面的參數(shù)進(jìn)行處理，處理完之后將返回的Observable交給下游來訂閱。

// 下面的map函數(shù)就是上面寫的那個(gè)，這是以前的寫法，現(xiàn)在不支持，
import {Observable} from 'rxjs/Observable'; 
import 'rxjs/add/observable/of'; 
import 'rxjs/add/operator/map';
import 'rxjs/add/operator/let'; 
const source$ = Observable.of(1, 2, 3); 
// 雖然map的使?是通過給Observable打補(bǔ)丁導(dǎo)?的，
// 但是map是直接作?于參數(shù)obs$，?不是作?于this，
// 所以，double$是?個(gè)純函數(shù)。
const double$ = obs$ => obs$.map(x => x * 2); 
// let的參數(shù)是?個(gè)函數(shù)，在這個(gè)例?中函數(shù)參數(shù)名為double$，
// 這個(gè)函數(shù)名也以$為后綴，代表它返回的是?個(gè)Observable對(duì)象，
// double$同樣接受?個(gè)Observable對(duì)象作為輸?參數(shù)，也就是說，
// double$的功能就是根據(jù)?個(gè)Observable對(duì)象產(chǎn)??個(gè)新的
// Observable對(duì)象。
const result$ = source$.let(double$);

過程：?let起到連接上游下游的作用，真正的工作完全由函數(shù)參數(shù)map來執(zhí)行。

處理之前的map函數(shù)：?此時(shí)map的實(shí)現(xiàn)部分也看不到對(duì)this的訪問，而是用一個(gè)參數(shù)obs$代替了 this，這樣，在數(shù)據(jù)管道中上游的Observable是以參數(shù)形式傳遞，而不是靠 this來獲得，讓map徹底成了一個(gè)純函數(shù)。map執(zhí)行返回的結(jié)果是一個(gè)函數(shù)，接受一個(gè)Observable對(duì)象返回一個(gè) Observable 對(duì)象，正好滿足let的參數(shù)要求。

const map = (fn) => (obs$) =>
 ?new Observable((observer) =>
 ? ?obs$.subscribe({
 ? ? ?next: (value) => observer.next(fn(value)),
 ? ? ?error: (err) => observer.error(error),
 ? ? ?complete: () => observer.complete(),
 ? ?})
 ?);

好處：?由于每一個(gè)lettable操作符都是純函數(shù)，而且也不會(huì)被作為補(bǔ)丁掛在Observable上，Tree Shaking就能夠找到根本不會(huì)被使用的操作符。

=> pipe操作符 <=

原因：?要導(dǎo)入let這個(gè)操作符，又不得不用傳統(tǒng)的打補(bǔ)丁或者使用call的方式，使用起來要導(dǎo)入很麻煩；所以創(chuàng)建了pipe操作符，它可以滿足let具備的功能。使用pipe只需像使用let那樣導(dǎo)入模塊，任何Observable對(duì)象都保持pipe，此外還有管道功能，可以把多個(gè)lettable操作符串接起來，形成數(shù)據(jù)管道。

import { map, filter, of } from "rxjs";
const source$ = of(1, 2, 3);
// 可以一次使用多個(gè)操作符
const result$ = source$.pipe(
 ?filter((x) => x % 2 === 0),
 ?map((x) => x * 2)
);
result$.subscribe(console.log);

三、創(chuàng)建數(shù)據(jù)流

（1）創(chuàng)建類操作符

說明：?這里所說的創(chuàng)造，是指這些操作符不依賴于其他Observable對(duì)象，這些操作符可以憑空或者根據(jù)其他數(shù)據(jù)源創(chuàng)造出?個(gè)Observable對(duì)象，其次創(chuàng)建類操作符往往不會(huì)從其他Observable對(duì)象獲取數(shù)據(jù)，因?yàn)樵跀?shù)據(jù)管道中它自己就是數(shù)據(jù)流的源頭，基于這些特性大部分的創(chuàng)建類操作符都是靜態(tài)操作符。

（2）創(chuàng)建同步數(shù)據(jù)流

說明：?對(duì)于同步的Observable對(duì)象，需要關(guān)心的是存在哪些數(shù)據(jù)和數(shù)據(jù)之間的先后順序，由于數(shù)據(jù)之間的時(shí)間間隔不存在因此不需要考慮時(shí)間方面的問題。

=> of操作符 <=

作用：?可以輕松創(chuàng)建指定數(shù)據(jù)集合的Observable對(duì)象；

參數(shù)：?of(數(shù)據(jù)1,數(shù)據(jù)2,數(shù)據(jù)3...)；

注意：?of操作符所產(chǎn)生的Observable對(duì)象被訂閱后會(huì)將參數(shù)依次吐出來，吐完之后會(huì)調(diào)用complete方法；吐的這個(gè)過程是同步的，也就是所有數(shù)據(jù)之間是不存在間隔的。

const { of } = Rx;
of(1).pipe();

const { of } = Rx;
of(1, 2, 3).pipe();

=> range操作符 <=

作用：?對(duì)需要產(chǎn)生多個(gè)很長連續(xù)數(shù)字序列的場景，就是得上range這個(gè)操作符了，range的含義就是“范圍”，只需要指定一個(gè)范圍的開始值和長度，range 就能夠產(chǎn)生這個(gè)范圍內(nèi)的依次+1的數(shù)字序列；同樣數(shù)據(jù)吐完之后會(huì)調(diào)用complete方法。

參數(shù)：?range(序列開始的任意數(shù)字,序列的長度)

const { range } = Rx;
range(1, 100).pipe();

局限性：?無法規(guī)定每次遞增的大小

=> generate操作符 <=

作用：?類似一個(gè)for循環(huán)，設(shè)定一個(gè)初始值，每次遞增這個(gè)值，直到滿足某個(gè)條件的時(shí)候才中止循環(huán)，同時(shí)，循環(huán)體內(nèi)可以根據(jù)當(dāng)前值產(chǎn)生數(shù)據(jù)。

參數(shù)：?generate(初始值, 條件判斷函數(shù), 值如何增加函數(shù), 返回結(jié)果處理函數(shù))

// 假設(shè)存在這樣的for循環(huán)：產(chǎn)??個(gè)?10?的所有偶數(shù)的平?
const result = [];
for (let i = 2; i < 10; i += 2) {
 ?result.push(i * i);
}
// 使用generate類似實(shí)現(xiàn)
const { generate } = Rx;
generate(
 ? ?2, // 初始值，相當(dāng)于for循環(huán)中的i=2
 ? ?value => value < 10, //繼續(xù)的條件，相當(dāng)于for中的條件判斷
 ? ?value => value + 2, //每次值的遞增
 ? ?value => value * value // 產(chǎn)?的結(jié)果
).pipe();

注意：?使用時(shí)需要保證后面三個(gè)函數(shù)參數(shù)為純函數(shù)

=> repeat操作符 <=

作用：?重復(fù)上游Observable中的數(shù)據(jù)n次

參數(shù)1：?repeat(重復(fù)的次數(shù))

const { of } = Rx;
const { repeat } = RxOperators;
of(1, 2, 3).pipe(repeat(100))

參數(shù)2：?repeat({count: 重復(fù)的次數(shù), delay: 數(shù)據(jù)的時(shí)間間隔})

import { of, repeat } from "rxjs";
of(1, 2, 3)
 ?.pipe(
 ? ?repeat({
 ? ? ?count: 10,
 ? ? ?delay: 1000,
 ? ?})
 ?)
 ?.subscribe((res) => console.log(res));

注意：?保證上游Observable對(duì)象一定會(huì)完結(jié)。

=> EMPTY常量 <=

作用：?產(chǎn)生一個(gè)直接完結(jié)的Observable對(duì)象，沒有參數(shù)，不產(chǎn)生任何數(shù)據(jù)，直接完結(jié)。

const { EMPTY ?} = Rx;
EMPTY.pipe()

=> throwError操作符 <=

作用：?它所產(chǎn)生的Observable對(duì)象也是什么都不做，直接出錯(cuò)，拋出的錯(cuò)誤就是throw的參數(shù)

參數(shù)：?throwError(錯(cuò)誤程序)

const { throwError ?} = Rx;
throwError(new Error('這是一個(gè)錯(cuò)誤')).pipe()

=> NEVER常量 <=

作用：?產(chǎn)生的Observable對(duì)象就真的是什么都不做，既不吐出數(shù)據(jù)，也不完結(jié)，也不產(chǎn)生錯(cuò)誤，就這樣待著，一直到永遠(yuǎn)

const { NEVER ?} = Rx;
NEVER.pipe()

（3）創(chuàng)建異步數(shù)據(jù)流

說明：?就是創(chuàng)建異步的Observable對(duì)象，不光要考慮產(chǎn)生什么數(shù)據(jù)，還需要考慮數(shù)據(jù)之間的時(shí)間間隔了

=> interval操作符 <=

作用：?定時(shí)從Observable對(duì)象吐出一個(gè)數(shù)據(jù)，如果不主動(dòng)結(jié)束就一直執(zhí)行

參數(shù)：?interval(吐數(shù)據(jù)的間隔毫秒數(shù))

const { interval } = Rx;
interval(1000).pipe()

注意：
它所創(chuàng)建的數(shù)據(jù)流不會(huì)自動(dòng)完結(jié)，也就是不會(huì)調(diào)用complete方法，要想結(jié)束只能夠執(zhí)行退訂操作了 其次這個(gè)異步數(shù)據(jù)序列總是從0開始遞增的； 最后它與原生的setinterval的地位是等價(jià)的

=> timer操作符 <=

作用：?產(chǎn)生的Observable對(duì)象在指定毫秒之后會(huì)吐出一個(gè)數(shù)據(jù)，執(zhí)行后立即結(jié)束

參數(shù)：?timer(毫秒數(shù) / 一個(gè)Date對(duì)象, 時(shí)間間隔)

// 明確延時(shí)產(chǎn)?數(shù)據(jù)的時(shí)間間隔
const { timer } = Rx;
timer(1000).pipe()

// 明確的是?個(gè)時(shí)間點(diǎn)
const { timer } = Rx;
timer(
 ?new Date(
 ? ?new Date().getTime() + 1000
 ?)
).pipe()

const { timer } = Rx;
timer(1000,2000).pipe()

注意：
如果使用第二個(gè)參數(shù)，產(chǎn)生的數(shù)據(jù)流跟interval類似，只不過產(chǎn)生第一個(gè)數(shù)據(jù)的時(shí)間間隔由第一個(gè)參數(shù)決定，后面產(chǎn)生數(shù)據(jù)的時(shí)間間隔由第二個(gè)參數(shù)決定；如果兩個(gè)參數(shù)一致，那就是interval了

=> from操作符 <=

作用：?以把任何可迭代對(duì)象都轉(zhuǎn)化為Observable對(duì)象

參數(shù)：?from(任何可迭代對(duì)象)

const { from } = Rx;
from([1,2,3]).pipe()

const { from } = Rx;
from('abc').pipe()

注意：?在from的眼中，把輸出參數(shù)都當(dāng)做一個(gè)Iterable來看待，所以上面的字符串a(chǎn)bc在from看來就和數(shù)組['a'，'b'，'c']沒有區(qū)別

import { from } from "rxjs";
const promise = Promise.resolve("good");
const source$ = from(promise);
source$.subscribe(
 ?console.log,
 ?(error) => console.log("catch", error),
 ?() => console.log("complete")
);

import { from } from "rxjs";
const promise = Promise.reject("error");
const source$ = from(promise);
source$.subscribe(
 ?console.log,
 ?(error) => console.log("catch", error),
 ?() => console.log("complete")
);

解釋：?如果from的參數(shù)是promise，當(dāng)promsie成功結(jié)束，from產(chǎn)生的Observable對(duì)象就會(huì)吐出Promise成功的結(jié)果，并且立刻結(jié)束，如果以失敗而告終的時(shí)候，from產(chǎn)生的Observable對(duì)象也會(huì)立刻產(chǎn)生失敗事件

=> fromEvent操作符 <=

作用1：?在網(wǎng)頁開發(fā)中，可以把DOM中的事件轉(zhuǎn)化為Observable對(duì)象中的數(shù)據(jù)

參數(shù)1：?fromEvent(事件源, 事件名稱)

// 希望點(diǎn)擊id為clickMe的按鈕時(shí)，id為text的div中的數(shù)字會(huì)增加1， 
// 連續(xù)點(diǎn)擊那個(gè)按鈕，對(duì)應(yīng)數(shù)字會(huì)持續(xù)增加
<template>
 ?<div>
 ? ?<button id="clickMe">Click Me</button>
 ? ?<div id="text">0</div>
 ?</div>
</template>
<script setup>
import { fromEvent } from "rxjs";
import { onMounted } from "vue";
let clickCount = 0;
onMounted(() => {
 ?const event$ = fromEvent(
 ? ? ?document.querySelector("#clickMe"), 
 ? ? ?"click"
 ?);
 ?event$.subscribe(() => {
 ? ?document
 ? ? ? ?.querySelector("#text")
 ? ? ? ?.innerText = ++clickCount;
 ?});
});
</script>
<style></style>

說明：?網(wǎng)頁開發(fā)中事件源一般是DOM節(jié)點(diǎn)

// 這里展示從Node.js的events中獲得數(shù)據(jù)
import { fromEvent } from "rxjs";
// 這個(gè)模塊需要使用 npm i events 安裝一下
import EventEmitter from "events";
const emitter = new EventEmitter();
// 只接受數(shù)據(jù)源中事件為"msg"的數(shù)據(jù)
const source$ = fromEvent(emitter, "msg");
source$.subscribe(
 ?console.log,
 ?(error) => console.log("catch", error),
 ?() => console.log("complete")
);
// emitter的emit函數(shù)發(fā)送任何名稱的事件，
// 第?個(gè)參數(shù)就是事件名稱，第?個(gè)參數(shù)是數(shù)據(jù)
emitter.emit("msg", 1);
emitter.emit("msg", 2);
emitter.emit("another-msg", "oops");
emitter.emit("msg", 3);

注意：?fromEvent產(chǎn)生的是Hot Observable，也就是數(shù)據(jù)的產(chǎn)生和訂閱是無關(guān)的，如果在訂閱之前調(diào)用emitter.emit，那有沒有Observer這些數(shù)據(jù)都會(huì)立刻吐出來，等不到訂閱的時(shí)候，當(dāng)添加了Observer的時(shí)候，仍然什么數(shù)據(jù)都獲得不到。

import { fromEvent } from "rxjs";
import EventEmitter from "events";
const emitter = new EventEmitter();
const source$ = fromEvent(emitter, "msg");
// 在訂閱之前發(fā)射數(shù)據(jù)
emitter.emit("msg", 1);
emitter.emit("msg", 2);
emitter.emit("another-msg", "oops");
emitter.emit("msg", 3);
source$.subscribe(
 ?console.log,
 ?(error) => console.log("catch", error),
 ?() => console.log("complete")
);

=> fromEventPattern操作符 <=

作用：?用于處理的Observable對(duì)象被訂閱和退訂時(shí)的動(dòng)作

參數(shù)：?fromEventPattern(被訂閱時(shí)觸發(fā)的函數(shù), 被退訂時(shí)觸發(fā)的函數(shù))

import { fromEventPattern } from "rxjs";
import EventEmitter from "events";
const emitter = new EventEmitter();
// handler參數(shù)可以理解為觀察者對(duì)象中的next方法
const addHandler = (handler) => {
 ?// 監(jiān)聽事件源中的msg事件，每次觸發(fā)事件執(zhí)行next方法
 ?emitter.addListener("msg", handler);
};
const removeHandler = (handler) => {
 ?// 與上面相反，會(huì)移除msg事件上面的next方法
 ?emitter.removeListener("msg", handler);
};
const source$ = fromEventPattern(addHandler, removeHandler);
const subscription = source$.subscribe(
 ?console.log,
 ?(error) => console.log("catch", error),
 ?() => console.log("complete")
);
emitter.emit("msg", "hello");
emitter.emit("msg", "world");
// 取消訂閱后emitter上面監(jiān)聽的事件被取消掉，
// 所以此處的值并不會(huì)出現(xiàn)在Observable對(duì)象里面
subscription.unsubscribe();
emitter.emit("msg", "end");

說明：?它提供的就是一種模式，不管數(shù)據(jù)源是怎樣的行為，最后的產(chǎn)出都是一個(gè)Observable對(duì)象

=> ajax操作符 <=

作用：?用于發(fā)送請(qǐng)求并根據(jù)結(jié)果返回Observable對(duì)象

參數(shù)：?ajax('請(qǐng)求的地址')

// 根據(jù)github上的api獲取RxJS項(xiàng)?獲得的Start的數(shù)量
<template>
 ?<div>
 ? ?<button id="getStar">Get RxJS Star Count</button>
 ? ?<div id="text"></div>
 ?</div>
</template>
<script setup>
import { onMounted } from "vue";
import { fromEvent } from "rxjs";
import { ajax } from "rxjs/ajax";
onMounted(() => {
 ?fromEvent(
 ? ? ?document.querySelector("#getStar"),
 ? ? ?"click"
 ?).subscribe(
 ? ? ?() => {
 ? ? ? ? ?ajax("https://api.github.com/repos/ReactiveX/rxjs")
 ? ? ? ? ?.subscribe(
 ? ? ? ? ? ? ?(value) => {
 ? ? ? ? ? ? ? ? ?const starCount = 
 ? ? ? ? ? ? ? ? ? ? ?value.response.stargazers_count;
 ? ? ? ? ? ? ? ? ? ? ?
 ? ? ? ? ? ? ? ? ?document.querySelector("#text").innerText = 
 ? ? ? ? ? ? ? ? ? ? ?starCount;
 ? ? ? ? ? });
 ?});
});
</script>

=> defer操作符 <=

作用：?用于延遲執(zhí)行某些操作

參數(shù)：?defer(一個(gè)函數(shù)，這個(gè)函數(shù)會(huì)在被訂閱時(shí)調(diào)用)

// 延遲發(fā)送請(qǐng)求
import { defer } from "rxjs";
import { ajax } from "rxjs/ajax";
defer(
 ? ?() => ajax("https://api.github.com/repos/ReactiveX/rxjs")
 ? ? ? ? ? ?.subscribe(
 ? ? ? ? ? ? ? ?(res) => console.log(res)
 ? ? ? ? ? ?)
);

四、合并數(shù)據(jù)流

（1）合并類操作符

說明：?其作用在于將有多個(gè)Observable對(duì)象作為數(shù)據(jù)來源，把不同來源的數(shù)據(jù)根據(jù)不同的規(guī)則合并到一個(gè)Observable對(duì)象中。

=> concat操作符 <=

作用：?把多個(gè)Observable中的數(shù)據(jù)內(nèi)容依次合并，合并的時(shí)候原有數(shù)據(jù)不變

參數(shù)：?concat(數(shù)據(jù)1, 數(shù)據(jù)2, 數(shù)據(jù)3...)

import { concat, of } from "rxjs";
const source1$ = of(1, 2, 3);
const source2$ = of(4, 5, 6);
concat(source1$, source2$)
 ? ?.subscribe(
 ? ? ? ?(res) => console.log(res)
 ? ?);

注意：?它在工作的時(shí)候，會(huì)先從第一個(gè)Observable對(duì)象中獲取數(shù)據(jù)，等它c(diǎn)omplete之后，再從下一個(gè)對(duì)象中去數(shù)據(jù)，直到取完所有的，此時(shí)，如果其中有一個(gè)對(duì)象是不完結(jié)的狀態(tài)，那么它之后的Observable對(duì)象就不會(huì)有被取到的機(jī)會(huì)。

=> merge操作符 <=

作用：?一定性訂閱上游所有的Observable對(duì)象，只要有數(shù)據(jù)傳遞下來，這個(gè)數(shù)據(jù)就會(huì)被傳遞給下游，也就是數(shù)據(jù)采取先到先出的原則，同時(shí)合并的時(shí)候原有數(shù)據(jù)不變

參數(shù)：?merge(數(shù)據(jù)1, 數(shù)據(jù)2, 數(shù)據(jù)3, ... 可選數(shù)字參數(shù))

場景一：合并異步數(shù)據(jù)流

import { merge, of, repeat, pipe } from "rxjs";
// 隔700ms重復(fù)一個(gè)A，重復(fù)的次數(shù)為5次
const source1$ = of("A").pipe(
 ? ? ? ? ? ? ? ? ? ? ? ?repeat(
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 700 }
 ? ? ? ? ? ? ? ? ? ? ? ?));
// 隔800ms重復(fù)一個(gè)B，重復(fù)的次數(shù)為5次
const source2$ = of("B").pipe(
 ? ? ? ? ? ? ? ? ? ? ? ?repeat(
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 800 }
 ? ? ? ? ? ? ? ? ? ? ? ?));
const merged$ = merge(source1$, source2$);
merged$.subscribe((res) => console.log(res));

場景二： 同步限流

解釋：?此時(shí)需要用到最后的參數(shù)，參數(shù)是一個(gè)數(shù)字，表示可以同時(shí)合并的個(gè)數(shù)

import { merge, of, repeat, pipe } from "rxjs";
// 隔700ms重復(fù)一個(gè)A，重復(fù)的次數(shù)為5次
const source1$ = of("A").pipe(
 ? ? ? ? ? ? ? ? ? ? ? ?repeat(
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 700 }
 ? ? ? ? ? ? ? ? ? ? ? ?));
// 隔800ms重復(fù)一個(gè)B，重復(fù)的次數(shù)為5次
const source2$ = of("B").pipe(
 ? ? ? ? ? ? ? ? ? ? ? ?repeat(
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 800 }
 ? ? ? ? ? ? ? ? ? ? ? ?));
 ? ? ? ? ? ? ? ? ? ? ? ?
// 隔900ms重復(fù)一個(gè)C，重復(fù)的次數(shù)為5次
const source3$ = of("C").pipe(
 ? ? ? ? ? ? ? ? ? ? ? ?repeat(
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 900 }
 ? ? ? ? ? ? ? ? ? ? ? ?));
 ? ? ? ? ? ? ? ? ? ? ? ?
// 限定合并的個(gè)數(shù)為2
const merged$ = merge(source1$, source2$, source3$, 2);
merged$.subscribe((res) => console.log(res));

場景三：合并多個(gè)事件

說明：?一個(gè)元素存在click事件和touch事件，對(duì)應(yīng)網(wǎng)頁和移動(dòng)設(shè)備，假如其事件處理程序的邏輯一致，此時(shí)就可以分別使用fromEvent獲取單個(gè)事件流，之后用merge合并成一個(gè)數(shù)據(jù)流，就可以集中管理了

// 可以像這樣處理
const click$ = fromEvent(element, 'click'); 
const touchend$ = fromEvent(element, 'touchend'); 
merge(click$, touchend$).subscribe(處理函數(shù));

=> zip操作符 <=

作用：?將數(shù)據(jù)流里面的數(shù)據(jù)一一對(duì)應(yīng)并使用數(shù)組組合起來

參數(shù)：?zip(數(shù)據(jù)流1, 數(shù)據(jù)流2, 數(shù)據(jù)流3...)

場景一： 一對(duì)一合并

import { interval, of, zip } from "rxjs";
// 一個(gè)異步數(shù)據(jù)流，產(chǎn)生的數(shù)據(jù)是無限的
const source1$ = interval(1000);
// 一個(gè)同步數(shù)據(jù)流，產(chǎn)生的數(shù)據(jù)流是有限
const source2$ = of("a", "b", "c");
// 將兩個(gè)數(shù)據(jù)流合并
zip(source1$, source2$)
 ? ?.subscribe(
 ? ? ? ?(res) => console.log(res),
 ? ? ? ?null,
 ? ? ? ?() => console.log('complete')
 ? ?);

注意：?這里數(shù)據(jù)的匹配是一一對(duì)應(yīng)的，所以數(shù)據(jù)少的那個(gè)Observable決定zip產(chǎn)生數(shù)據(jù)的個(gè)數(shù)；然后在對(duì)應(yīng)的時(shí)候需要雙方都有數(shù)據(jù)才能夠?qū)?yīng)，這也是為什么上面的打印會(huì)隔1s才打印。

問題： 數(shù)據(jù)積壓

說明：?如果某個(gè)上游source1$吐出數(shù)據(jù)的速度很快，而另一個(gè)上游source2$吐出數(shù)據(jù)的速度很慢，那zip就不得不先存儲(chǔ)source1$吐出的數(shù)據(jù)，因?yàn)镽xJS的工作方式是“推”， Observable把數(shù)據(jù)推給下游之后就沒有責(zé)任保存數(shù)據(jù)了。被source1$推送了數(shù)據(jù)之后，zip就有責(zé)任保存這些數(shù)據(jù)，等著和source2$未來吐出的數(shù)據(jù)配對(duì)。假如source2$遲遲不吐出數(shù)據(jù)，那么zip就會(huì)一直保存source1$沒有配對(duì)的數(shù)據(jù)，然而這時(shí)候source1$可能會(huì)持續(xù)地產(chǎn)生數(shù)據(jù)，最后zip積壓的數(shù)據(jù)就會(huì)越來越多，占用的內(nèi)存也就越來越多。對(duì)于數(shù)據(jù)量比較小的Observable對(duì)象，這樣的數(shù)據(jù)積壓還可以忍受，但是對(duì)于超大量的數(shù)據(jù)流，使用zip就不得不考慮潛在的內(nèi)存壓力問題。

=> combineLatest操作符 <=

作用：?合并上游所有Observable一個(gè)最新的數(shù)據(jù)，也就是它返回值是一個(gè)數(shù)組

參數(shù)：?combineLatest([數(shù)據(jù)1, 數(shù)據(jù)2, 數(shù)據(jù)3 ...], 處理函數(shù))

場景一： 基本使用

import { combineLatest, timer } from "rxjs";
// 隔1s產(chǎn)生一個(gè)數(shù)字
const firstTimer = timer(1000, 1000);
// 隔2s產(chǎn)生一個(gè)數(shù)字
const secondTimer = timer(1000, 2000);
// 合并上面的數(shù)據(jù)流
const combinedTimers = combineLatest([firstTimer, secondTimer]);
combinedTimers.subscribe((value) => console.log(value));

注意：?首先還是一一對(duì)應(yīng)的關(guān)系，也就是如果一個(gè)數(shù)據(jù)源還沒發(fā)射值出來，那么會(huì)等待它將值發(fā)射出來，如果值沒有改變并且操作沒有完結(jié)的話，發(fā)射的值將一直是這一個(gè)，所以只有所有的Observable對(duì)象完結(jié)，combineLatest才會(huì)給下游一個(gè)complete信號(hào)，表示不會(huì)有任何數(shù)據(jù)更新了

場景二： 合并同步數(shù)據(jù)流

const firstTimer = of("a", "b", "c");
const secondTimer = of(1, 2, 3);
const combinedTimers = combineLatest([firstTimer, secondTimer]);
combinedTimers.subscribe((value) => console.log(value));

工作方式：?combineLatest在工作的時(shí)候，會(huì)按照順序依次訂閱所有上游的Observable對(duì)象，只有所有上游Observable對(duì)象都已經(jīng)吐出數(shù)據(jù)了，才會(huì)給下游傳遞所有上游“最新數(shù)據(jù)”組合的數(shù)據(jù)

解釋：?由于of產(chǎn)生的同步數(shù)據(jù)流，在被訂閱時(shí)就會(huì)吐出所有數(shù)據(jù)，最后一個(gè)吐出的數(shù)據(jù)是字符串c，這也就是最新的數(shù)據(jù)，然后訂閱下一個(gè)對(duì)象，下一個(gè)對(duì)象會(huì)依次吐出數(shù)據(jù)，然后跟上一個(gè)對(duì)象的最新數(shù)據(jù)c結(jié)合，這就得到了上面看到的內(nèi)容

場景三：定制下游數(shù)據(jù)

說明：?這里就需要啊使用處理函數(shù)了，這個(gè)函數(shù)的參數(shù)就是每一個(gè)數(shù)據(jù)源的最新值，其返回值就是下游所接受到的數(shù)據(jù)，如果沒有返回值，則下游收到的數(shù)據(jù)為undefined

import { combineLatest, timer, of } from "rxjs";
const firstTimer = of("a", "b", "c");
const secondTimer = of(1, 2, 3);
const combinedTimers = combineLatest(
 ?[firstTimer, secondTimer],
 ?(res1, res2, res3) => {
 ? ?// 上面只有兩個(gè)數(shù)據(jù)源，所以參數(shù)只會(huì)前兩個(gè)有值
 ? ?console.log(res1, res2, res3);
 ?}
);
combinedTimers.subscribe();

=> withLatestFrom操作符 <=

說明：?這個(gè)的作用于combineLatest是類似的，只不過下游推送數(shù)據(jù)只能由一個(gè)上游Observable對(duì)象驅(qū)動(dòng)，也就是調(diào)用withLatestFrom的那個(gè)Observable對(duì)象起到主導(dǎo)數(shù)據(jù)產(chǎn)生節(jié)奏的作用，作為參數(shù)的Observable對(duì)象只能貢獻(xiàn)數(shù)據(jù)，不能控制產(chǎn)生數(shù)據(jù)的時(shí)機(jī)

參數(shù)：?數(shù)據(jù)源1.withLatestFrom(數(shù)據(jù)源2)

import { withLatestFrom, timer, pipe, map } from "rxjs";
// 每隔兩秒產(chǎn)生100、200、300這樣的數(shù)字
const source1$ = timer(0, 2000)
 ? ? ? ? ? ? ? ? ? ?.pipe(
 ? ? ? ? ? ? ? ? ? ? ? ?map((x) => 100 * x)
 ? ? ? ? ? ? ? ? ? ?);
// 每隔一秒產(chǎn)生0、1、2這樣的數(shù)字
const source2$ = timer(500, 1000);
// 后面的處理函數(shù)將它們想加起來
const result$ = source1$
 ? ? ? ? ? ? ? ? ? ?.pipe(
 ? ? ? ? ? ? ? ? ? ? ? ?withLatestFrom(
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?source2$, 
 ? ? ? ? ? ? ? ? ? ? ? ? ? ?(a, b) => a + b
 ? ? ? ? ? ? ? ? ? ? ? ?)
 ? ? ? ? ? ? ? ? ? ?);
result$.subscribe(console.log);

=> race操作符 <=

作用：?以O(shè)bservable產(chǎn)生第一個(gè)數(shù)據(jù)的時(shí)間為準(zhǔn)，只留下最快的那一個(gè)，當(dāng)然，使用的所有數(shù)據(jù)也是最快的那一個(gè)

參數(shù)：?race(數(shù)據(jù)源1, 數(shù)據(jù)源2, 數(shù)據(jù)源3 ...)

import { timer, race, pipe, map } from "rxjs";
// 立即開始產(chǎn)生數(shù)據(jù)a
const source1$ = timer(0, 2000).pipe(map(() => "a"));
// 500ms后開始產(chǎn)生數(shù)據(jù)b
const source2$ = timer(500, 1000).pipe(map(() => "b"));
// 比賽
const winner$ = race(source1$, source2$);
winner$.subscribe(console.log);

=> startWith操作符 <=

作用：?在讓?個(gè)Observable對(duì)象在被訂閱的時(shí)候，總是先同步吐出指定的若?個(gè)數(shù)據(jù)

參數(shù)：?數(shù)據(jù)源.startWith(參數(shù)1, 參數(shù)2, 參數(shù)3 ...)

import { of, startWith } from "rxjs";
of(1000)
 ?.pipe(startWith("timer start", 1, 2))
 ?.subscribe((x) => console.log(x));

=> forkJoin操作符 <=

作用：?等待所有參數(shù)Observable對(duì)象的最后?個(gè)數(shù)據(jù)，將其合并成一個(gè)數(shù)組發(fā)射出去

參數(shù)：?forkJoin(對(duì)象 / 數(shù)組)

import { forkJoin, of, timer } from "rxjs";
// 下面會(huì)在四秒后返回結(jié)果
forkJoin({
 ?foo: of(1, 2, 3, 4),
 ?bar: Promise.resolve(8),
 ?baz: timer(4000),
}).subscribe((res) => console.log(res));

（2）高階Observable

說明：?簡單理解就是一個(gè)Observable中存在Observable，它有一個(gè)特點(diǎn)就是高階Observable完結(jié)不代表其里面的Observable完結(jié)

=> concatAll操作符 <=

說明：?這個(gè)操作符針對(duì)高階Observable，也是依次訂閱Observable內(nèi)部的Observable取值結(jié)合，訂閱的過程中如果上一個(gè)Observable沒有完結(jié)就不會(huì)訂閱下一個(gè)Observable對(duì)象。其他操作可以參照concat

參數(shù)：?沒有參數(shù)

import { of, concatAll } from "rxjs";
const source = of(
 ? ? ? ? ? ? ? ? ? ?of(1, 2, 3),
 ? ? ? ? ? ? ? ? ? ?of(4, 5, 6),
 ? ? ? ? ? ? ? ? ? ?of(7, 8, 9)
 ? ? ? ? ? ? ? ?);
source.pipe(concatAll())
 ? ? ? ?.subscribe(
 ? ? ? ? ? ?(val) => console.log(val)
 ? ? ? ?);

=> mergeAll操作符 <=

說明：?針對(duì)高階Observable，在合并的時(shí)候，依次訂閱其內(nèi)部的Observable對(duì)象，那個(gè)對(duì)象有數(shù)據(jù)傳下來，這個(gè)數(shù)據(jù)就會(huì)傳遞給下游；它可以傳遞一個(gè)數(shù)字來限定合并的最大流的個(gè)數(shù)。其他操作可以參照merge

參數(shù)：?mergeAll(數(shù)字)

import { of, mergeAll, repeat } from "rxjs";
// 這里A延遲復(fù)制的時(shí)間比B長，所以第二次打印的時(shí)候B在前面
const source = of(
 ?of("A").pipe(
 ? ? ? ? ? ? ?repeat({ count: 5, delay: 800 })
 ? ? ? ? ?),
 ?of("B").pipe(
 ? ? ? ? ? ? ?repeat({ count: 5, delay: 700 })
 ? ? ? ? ?)
);
source.pipe(mergeAll())
 ? ? ? ? ? ?.subscribe(
 ? ? ? ? ? ? ? ?(val) => console.log(val)
 ? ? ? ? ? ?);

=> zipAll操作符 <=

說明：?對(duì)高階Observable使用的時(shí)候，將數(shù)據(jù)流里面的數(shù)據(jù)一一對(duì)應(yīng)并使用數(shù)組組合起來。其它操作可以參考zip

參數(shù)：?zipAll(處理函數(shù))

import { of, zipAll } from "rxjs";
const source = of(of("A", "B", "C"), of(1, 2, 3), of("X", "Y", "Z"));
source
 ?.pipe(
 ? ?// 可以接收一個(gè)處理函數(shù)，每個(gè)參數(shù)對(duì)應(yīng)返回值的每一項(xiàng)
 ? ?zipAll((a, b, c) => {
 ? ? ?// 這里將參數(shù)打印出來，如果沒有返回值則下游將收不到值
 ? ? ?console.log(a, b, c);
 ? ?})
 ?)
 ?.subscribe();

=> combineLatestAll操作符 <=

說明：?在處理高階Observable的時(shí)候，將其內(nèi)部Observable產(chǎn)生的最新數(shù)據(jù)一一對(duì)應(yīng)并用數(shù)組的形式返回出來。其它操作可以參考combineLatest

參數(shù)：?combineLatestAll(處理函數(shù))

import { of, combineLatestAll } from "rxjs";
const source = of(of("A", "B", "C"), of(1, 2, 3), of("X", "Y", "Z"));
source
 ?.pipe(
 ? ?// 可以接收一個(gè)處理函數(shù)，每個(gè)參數(shù)對(duì)應(yīng)返回值的每一項(xiàng)
 ? ?combineLatestAll((a, b, c) => {
 ? ? ?// 這里將參數(shù)打印出來，如果沒有返回值則下游將收不到值
 ? ? ?console.log(a, b, c);
 ? ?})
 ?)
 ?.subscribe();

五、輔助類操作符

（1）數(shù)學(xué)類操作符

說明：?這里介紹的操作符會(huì)遍歷上游Observable對(duì)象中吐出的所有數(shù)據(jù)才給下游傳遞數(shù)據(jù)， 也就是說，它們只有在上游完結(jié)的時(shí)候，才給下游傳遞唯?數(shù)據(jù)。

=> count操作符 <=

作用：?用于統(tǒng)計(jì)上游Observable對(duì)象吐出的所有數(shù)據(jù)個(gè)數(shù)，所以上游的Observable需要完結(jié)

參數(shù)：?count(過濾函數(shù))

import { of, interval, count } from "rxjs";
// 可以完結(jié)
of(1000, 1)
 ?.pipe(
 ? ? ?// 此時(shí)過濾出數(shù)據(jù)為1的數(shù)量
 ? ? ?count(
 ? ? ? ? ?(val) => val === 1
 ? ? ?)
 ?)
 ?.subscribe((res) => console.log(res));
// 無法完結(jié)
interval(1000)
 ?.pipe(count())
 ?.subscribe((res) => console.log(res));

=> max和min操作符 <=

作用：?找出上游數(shù)據(jù)中的最大值和最小值

參數(shù)：?max(比較函數(shù)) / min(比較函數(shù))

import { of, max } from "rxjs";
of(
 ? ?{ age: 7, name: "Foo" },
 ? ?{ age: 5, name: "Bar" },
 ? ?{ age: 9, name: "Beer" }
).pipe(
 ? ?// 返回值為正 => a > b
 ? ?// 返回值為0 => a = b
 ? ?// 返回值為負(fù) => a < b
 ? ?max((a, b) => a.age > b.age)
 ?)
 .subscribe(
 ? ? (x) => console.log(x.name)
 );

注意：?如果Observable吐出的數(shù)據(jù)類型是復(fù)雜數(shù)據(jù)類型，?如?個(gè)對(duì)象，那必須指定?個(gè)?較這種復(fù)雜類型??的?法，就像上面使用的那樣

=> reduce操作符 <=

說明：?對(duì)上游的每個(gè)數(shù)據(jù)進(jìn)行自定義計(jì)算，也就是對(duì)每一個(gè)元素都會(huì)調(diào)用一次這個(gè)函數(shù)

參數(shù)：?reduce((累加的值, 當(dāng)前元素的值) => {}, 初始值)

// 計(jì)算 1-100 的和
import { range, reduce } from "rxjs";
range(1, 100)
 ?.pipe(
 ? ? ?reduce(
 ? ? ? ? ?(acc, current) => acc + current,
 ? ? ? ? ?0
 ? ? ?)
 ?)
 ?.subscribe(
 ? ? ?(res) => console.log(res)
 ?);

2）條件操作符

說明：?根據(jù)上游Observable對(duì)象的某些條件產(chǎn)生一個(gè)新的 Observable對(duì)象

=> every操作符 <=

作用：?它接受一個(gè)判定函數(shù)作為參數(shù)，如果上游所有數(shù)據(jù)都能夠通過這個(gè)函數(shù)，那么會(huì)返回一個(gè)包含true值的Observable，有一個(gè)通不過就返回一個(gè)包含false值的Observable，在吐出結(jié)果后every產(chǎn)生的Observable會(huì)立即完結(jié)；不要對(duì)不會(huì)完結(jié)的對(duì)象使用

參數(shù)：?every(判定函數(shù))

import { every, of } from "rxjs";
of(1, 100)
 ?.pipe(
 ? ? ?every(
 ? ? ? ? ?// 這里判定是否所有值都大于10，顯然1不行
 ? ? ? ? ?(val) => val > 10
 ? ? ?)
 ?)
 ?.subscribe((res) => console.log(res));

=> find和findIndex操作符 <=

作用：?通過一個(gè)處理函數(shù)來在上游數(shù)據(jù)中查找滿足條件的數(shù)據(jù)，find會(huì)吐出找到的上游的數(shù)據(jù)，findIndex會(huì)吐出滿足判定條件的索引，如果找不到find會(huì)吐出undefined后完結(jié)，findIndex則會(huì)吐出-1后完結(jié)；不要對(duì)不會(huì)完結(jié)的對(duì)象使用

參數(shù)：?find(處理函數(shù)) / findIndex(處理函數(shù))

import { find, findIndex, of } from "rxjs";
of(1, 100)
 ?.pipe(
 ? ? ?find(
 ? ? ? ? ?(val) => (val = 100)
 ? ? ?)
 ?)
 ?.subscribe((res) => console.log(res));
 ?
of(1, 100)
 ?.pipe(
 ? ? ?findIndex(
 ? ? ? ? ?(val) => (val = 100)
 ? ? ?)
 ?)
 ?.subscribe((res) => console.log(res));

=> isEmpty操作符 <=

作用：?檢測上游Observable對(duì)象是不是空的，如果在完結(jié)之前沒有吐出數(shù)據(jù)，它就是空的，此時(shí)返回一個(gè)包含true值的Observable，否則返回一個(gè)包含false值的Observable

import { EMPTY, isEmpty, of } from "rxjs";
// 不是空的
of(1)
 ?.pipe(isEmpty())
 ?.subscribe((res) => console.log(res));
 ?
// 是空的
EMPTY.pipe(isEmpty())
 ?.subscribe((res) => console.log(res));

=> defaultIfEmpty操作符 <=

作用：?接受一個(gè)默認(rèn)值，如果檢測上游的Observable是空的，則把這個(gè)默認(rèn)值傳遞給下游，如果不是空的就把上游的東西傳遞給下游；如果不傳但是上游檢測還是空的，下游就會(huì)收到一個(gè)undefined值

import { defaultIfEmpty, EMPTY, of } from "rxjs";
// 不是空值，不傳參數(shù)
of(1)
 ?.pipe(defaultIfEmpty())
 ?.subscribe((res) => console.log(res));
// 不是空值，傳參數(shù)
of(1)
 ?.pipe(defaultIfEmpty("存在內(nèi)容"))
 ?.subscribe((res) => console.log(res));
// 是空值，不傳參數(shù)
EMPTY.pipe(defaultIfEmpty())
 ?.subscribe((res) => console.log(res));
 ?
// 是空值，傳參數(shù)
EMPTY.pipe(defaultIfEmpty("存在內(nèi)容"))
 ?.subscribe((res) => console.log(res));

六、過濾數(shù)據(jù)流

（1）過濾類操作符

說明：?對(duì)上游Observable中所有的數(shù)據(jù)使用判定函數(shù)進(jìn)行操作，決定是否某些元素不能通過進(jìn)入下游，如果對(duì)某個(gè)元素處理結(jié)果為true，表示能通過，否則就不能通過

=> filter操作符 <=

作用：?跟JavaScript中的filter使用起來是類似的，只不過這里針對(duì)的是Observable

參數(shù)：?filter(過濾函數(shù))

import { filter, interval } from "rxjs";
source$ = interval(1000)
 ?.pipe(
 ? ? ?// 過濾能被2整除的數(shù)據(jù)
 ? ? ?filter(
 ? ? ? ? ?(x) => x % 2 === 0
 ? ? ?)
 ?)
 ?
source$.subscribe((res) => console.log(res));

注意：?當(dāng)上游產(chǎn)?數(shù)據(jù)的時(shí)候，只要這個(gè)數(shù)據(jù)滿?判定條件，就會(huì)立刻被同步傳給下游。

=> first操作符 <=

作用：?過濾出Observable中第一個(gè)滿足條件的值，在沒有找到的時(shí)候會(huì)拋出一個(gè)錯(cuò)誤，如果不想這個(gè)錯(cuò)誤傳遞給下游可以使用第二個(gè)默認(rèn)值，它的作用是在沒找到滿足條件的值的時(shí)候?qū)⑦@個(gè)值傳遞出去。如果不傳參數(shù)則將第一個(gè)數(shù)據(jù)返回出去，

參數(shù)：?filter(過濾函數(shù), 默認(rèn)值)

import { first, of } from "rxjs";
// 找不到結(jié)果拋出錯(cuò)誤，但是給默認(rèn)值
of(1, 3)
 ?.pipe(first((x) => x % 2 === 0, 2))
 ?.subscribe((res) => console.log(res));
// 找到結(jié)果
of(1, 4, 3)
 ?.pipe(first((x) => x % 2 === 0))
 ?.subscribe((res) => console.log(res));
// 找不到結(jié)果拋出錯(cuò)誤
of(1, 3)
 ?.pipe(first((x) => x % 2 === 0))
 ?.subscribe((res) => console.log(res));

=> last操作符 <=

說明：?這個(gè)作用與first相反，它是找最后一個(gè)滿足條件的值，使用可以參考first，這里需要注意，使用這個(gè)操作符的上游必須完結(jié)，否則操作符不知道哪一個(gè)是最后一個(gè)數(shù)據(jù)

參數(shù)：?filter(過濾函數(shù), 默認(rèn)值)

import { last, interval } from "rxjs";
// 這個(gè)Observable不會(huì)完結(jié)，自然也不會(huì)拿到結(jié)果
interval(1000)
 ?.pipe(last((x) => x % 2 === 0, 2))
 ?.subscribe((res) => console.log(res));

=> take操作符 <=

作用：?從上游的數(shù)據(jù)中拿指定個(gè)數(shù)的數(shù)據(jù)，拿完之后就會(huì)完結(jié)，并將獲取的數(shù)據(jù)返回

參數(shù)：?take(需要的個(gè)數(shù))

import { interval, of, take } from "rxjs";
// 數(shù)據(jù)不夠拿，那就拿完為止
of("a", "b", "c")
 ?.pipe(take(4))
 ?.subscribe((res) => console.log(res));
// 獲取指定個(gè)數(shù)的數(shù)據(jù)
interval(1000)
 ?.pipe(take(4))
 ?.subscribe((res) => console.log(res));

注意：?上游每產(chǎn)生一個(gè)數(shù)據(jù)就會(huì)立即傳給下游，也就是同步操作的

=> takeLast操作符 <=

作用：?從后往前獲取指定個(gè)數(shù)的數(shù)據(jù)，之后將數(shù)據(jù)一次性返回出去之后完結(jié)

參數(shù)：?takeLast(需要的個(gè)數(shù))

import { interval, of, takeLast } from "rxjs";
// 數(shù)據(jù)不夠拿，那就拿完為止
of("a", "b", "c")
 ?.pipe(takeLast(4))
 ?.subscribe((res) => console.log(res));
// 數(shù)據(jù)沒有完結(jié)，獲取不到數(shù)據(jù)
interval(1000)
 ?.pipe(takeLast(4))
 ?.subscribe((res) => console.log(res));

注意：?如果上游的Observable對(duì)象不會(huì)完結(jié)的話，那么是拿不到數(shù)據(jù)的，因?yàn)椴恢勒l是最后一個(gè)數(shù)據(jù)

=> takeWhile操作符 <=

說明：?takeWhile接受?個(gè)判定函數(shù)作為參數(shù)，這個(gè)判定函數(shù)有兩個(gè)參數(shù)，分別代表上游的數(shù)據(jù)和對(duì)應(yīng)的序號(hào)，takeWhile會(huì)吐出上游數(shù)據(jù)，直到判定函數(shù)返回false，只要遇到第一個(gè)判定函數(shù)返回false的情況， takeWhile產(chǎn)生的Observable就完結(jié)

參數(shù)：?takeWhile(判定函數(shù), 布爾值)

// 這里關(guān)注第二個(gè)參數(shù)
import { range, takeWhile } from "rxjs";
range(1, 10)
 ?.pipe(
 ? ? ?takeWhile(
 ? ? ? ? ?(val) => val < 3, true
 ? ? ?)
 ?)
 ?.subscribe((res) => console.log(res));
range(1, 10)
 ?.pipe(
 ? ? ?takeWhile(
 ? ? ? ? ?(val) => val < 3, false
 ? ? ?)
 ?)
 ?.subscribe((res) => console.log(res));

注意：?第二個(gè)參數(shù)表示是否將第一次導(dǎo)致判定函數(shù)結(jié)果為false的那個(gè)值發(fā)射出去，默認(rèn)是false，表示不發(fā)射，true則表示發(fā)射。

=> takeUntil操作符 <=

說明：?它接受一個(gè)Observable對(duì)象，在這個(gè)對(duì)象沒有吐出數(shù)據(jù)之前，上游的數(shù)據(jù)會(huì)直接傳遞給下游，在參數(shù)對(duì)象吐出第一個(gè)數(shù)據(jù)時(shí)，上游的數(shù)據(jù)就不能傳遞給下游了。其次參數(shù)對(duì)象出現(xiàn)錯(cuò)誤的時(shí)候，這個(gè)錯(cuò)誤會(huì)傳遞給下游，此時(shí)上游數(shù)據(jù)也不能傳遞給下游了

參數(shù)：?takeUntil(Observable對(duì)象)

// 假如使用interval創(chuàng)建數(shù)據(jù)，在第三秒的時(shí)候停止
import { interval, takeUntil, timer } from "rxjs";
interval(1000)
 ?.pipe(
 ? ? ?takeUntil(timer(3000))
 ?)
 ?.subscribe((res) => console.log(res));

=> skip操作符 <=

作用：?跳過上游的前n個(gè)值，然后從上游的第n+1個(gè)值開始傳遞給下游，這個(gè)操作符不關(guān)心最后一個(gè)值是什么，所以這個(gè)操作符的上游不管會(huì)不會(huì)完結(jié)下游都會(huì)有值。

參數(shù)：?skip(跳過的個(gè)數(shù))

import { interval, skip } from "rxjs";
// 跳過前兩個(gè)值
interval(1000)
 ?.pipe(skip(2))
 ?.subscribe((res) => console.log(res));

=> skipLast操作符 <=

作用：?可以理解成去除上游的最后n個(gè)值，然后將剩下的值傳遞給下游；

參數(shù)：?skipLast(跳過的n個(gè)值)

import { interval, skipLast, of } from "rxjs";
// 一個(gè)完結(jié)的對(duì)象
of("a", "b", "c")
 ?.pipe(skipLast(2))
 ?.subscribe((res) => console.log(res));
// 不會(huì)完結(jié)的對(duì)象
interval(1000)
 ?.pipe(skipLast(2))
 ?.subscribe((res) => console.log(res));

注意：?上游沒有完結(jié)下游依然可以收到數(shù)據(jù)

=> skipWhile操作符 <=

說明：?它接收一個(gè)函數(shù)作為參數(shù)，上游的每一個(gè)數(shù)據(jù)都會(huì)執(zhí)行這個(gè)函數(shù)，只要有一個(gè)數(shù)據(jù)在函數(shù)中的返回值是false，那么這個(gè)數(shù)據(jù)之前的數(shù)據(jù)都會(huì)被過濾調(diào)用，剩下的數(shù)據(jù)會(huì)傳遞給下游。

參數(shù)：?skipWhile(處理函數(shù))

import { interval, skipWhile } from "rxjs";
interval(1000)
 ?.pipe(skipWhile((val) => val % 2 === 0))
 ?.subscribe((res) => console.log(res));

=> skipUntil操作符 <=

作用：?用于在一個(gè)Observable中跳過一些值，直到另一個(gè)Observable發(fā)出了特定的信號(hào)或者達(dá)到某種狀態(tài)。

參數(shù)：?skipUntil(Observable對(duì)象)

import { interval, timer, skipUntil } from "rxjs";
// 創(chuàng)建一個(gè)每秒發(fā)出一個(gè)值的Observable
const source$ = interval(1000);
// 創(chuàng)建一個(gè)在5秒后發(fā)出第一個(gè)值的Observable
const trigger$ = timer(5000);
// 使用skipUntil操作符，跳過source$的值，直到trigger$發(fā)出第一個(gè)值
const example$ = source$.pipe(skipUntil(trigger$));
const subscription = example$.subscribe((val) => console.log(val));

（2）有損回壓控制

解釋：?如果數(shù)據(jù)管道中某一個(gè)環(huán)節(jié)處理數(shù)據(jù)的速度跟不上數(shù)據(jù)涌現(xiàn)的速度，上游無法把數(shù)據(jù)推送給下游，就會(huì)在緩沖區(qū)中積壓數(shù)據(jù)，這就相當(dāng)于對(duì)上游施加了壓力，這就是RxJS世界中的回壓。

處理：?造成這種現(xiàn)象的原因是數(shù)據(jù)管道中某個(gè)環(huán)節(jié)數(shù)據(jù)涌?的速度超過了處理速度，那么，既然處理不過來，干脆就舍棄掉某些涌現(xiàn)的數(shù)據(jù)，這種方式稱為有損回壓控制

可選的調(diào)度器：
asyncScheduler：這是默認(rèn)的調(diào)度器，它使用setTimeout或setInterval來安排任務(wù)的執(zhí)行。它適用于異步操作。 queueScheduler：這個(gè)調(diào)度器會(huì)按順序執(zhí)行任務(wù)，并且會(huì)等待當(dāng)前任務(wù)完成后才執(zhí)行下一個(gè)任務(wù)。適用于同步操作。 animationFrameScheduler：這個(gè)調(diào)度器會(huì)根據(jù)瀏覽器的刷新率來執(zhí)行任務(wù)，通常用于實(shí)現(xiàn)動(dòng)畫效果或者對(duì)性能要求較高的操作。 asapScheduler：這個(gè)調(diào)度器會(huì)盡可能快地在當(dāng)前執(zhí)行棧中執(zhí)行任務(wù)，但是會(huì)在微任務(wù)隊(duì)列中等待其他任務(wù)完成后執(zhí)行。適用于需要盡快執(zhí)行的任務(wù)。 TestScheduler：這是用于測試的調(diào)度器，可以用來模擬時(shí)間的流逝，方便測試 RxJS 代碼。
可選參數(shù)對(duì)象：
leading：布爾值，表示是否在節(jié)流周期的開始時(shí)立即發(fā)出第一個(gè)值。默認(rèn)為true。 trailing：布爾值，表示是否在節(jié)流周期結(jié)束時(shí)發(fā)出最后一個(gè)值。默認(rèn)為false。

=> throttleTime操作符 <=

說明：?在一個(gè)時(shí)間范圍內(nèi)，上游傳遞給下游的數(shù)據(jù)只能傳遞一個(gè)；這里參數(shù)如果只傳一個(gè)，其它值都會(huì)使用默認(rèn)值；

參數(shù)：?throttleTime(時(shí)間范圍, 調(diào)度器, 可選參數(shù)對(duì)象)

import { 
 ? ? ? ?interval,
 ? ? ? ?throttleTime,
 ? ? ? ?asyncScheduler
} from "rxjs";
// 這里每隔1s產(chǎn)生一個(gè)數(shù)字
interval(1000)
 ?.pipe(
 ? ? ?throttleTime(
 ? ? ? ? ? ? ? ? ? ? ?2000,
 ? ? ? ? ? ? ? ? ? ? ?asyncScheduler,
 ? ? ? ? ? ? ? ? ? ? ?// trailing為true時(shí)產(chǎn)生的結(jié)果是：2、4、6...
 ? ? ? ? ? ? ? ? ? ? ?// leading為true時(shí)產(chǎn)生的結(jié)果是：3、6、9...
 ? ? ? ? ? ? ? ? ? ? ?{ leading: false, trailing: true }
 ? ? ? ? ? ? ? ? ?)
 ? ? ?)
 ?.subscribe((res) => console.log(res));

=> debounceTime操作符 <=

說明：?在一個(gè)時(shí)間范圍內(nèi)，一直有數(shù)據(jù)產(chǎn)生一直不會(huì)將數(shù)據(jù)傳遞給下游，只有在這個(gè)時(shí)間外產(chǎn)生的第一個(gè)數(shù)據(jù)才會(huì)傳遞給下游；所以產(chǎn)生數(shù)據(jù)的間隔需要大于這個(gè)時(shí)間范圍才可以

參數(shù)：?throttleTime(時(shí)間范圍, 調(diào)度器)

import { interval, debounceTime, asyncScheduler } from "rxjs";
// 這里的值如果比2000還小那么就不會(huì)有數(shù)據(jù)打印出來
interval(4000)
 ?.pipe(debounceTime(2000, asyncScheduler))
 ?.subscribe((res) => console.log(res));

=> throttle和debounce操作符 <=

作用：?這兩個(gè)都是使用Observable中的數(shù)據(jù)來控制流量，區(qū)別在于時(shí)機(jī)不同而已

參數(shù)：?throttle(處理函數(shù), 可選參數(shù)對(duì)象)

參數(shù)：?debounce(處理函數(shù))

// 這里以throttle為例
import { interval, timer, throttle } from "rxjs";
const source$ = interval(1000);
// 處理函數(shù)的參數(shù)只能拿到上游的數(shù)據(jù)
const durationSelector = (value) => {
 ?console.log(`# call durationSelector with ${value}`);
 ?return timer(2000);
};
const result$ = source$.pipe(throttle(durationSelector));
result$.subscribe(console.log);

理解：?當(dāng)source$產(chǎn)生第一個(gè)數(shù)據(jù)0的時(shí)候，throttle就和throttleTime一樣，毫不 猶豫地把這個(gè)數(shù)據(jù)0傳給了下游，在此之前會(huì)將這個(gè)數(shù)據(jù)0作為參數(shù)調(diào)用 durationSelector，然后訂閱durationSelector返回的Observable對(duì)象，在這個(gè) Observable對(duì)象產(chǎn)生第一個(gè)對(duì)象之前，所有上游傳過來的數(shù)據(jù)都會(huì)被丟棄，于是，source$產(chǎn)生的數(shù)據(jù)1就被丟棄了，因?yàn)閐urationSelector返回的 Observable對(duì)象被訂閱之后2000毫秒才會(huì)產(chǎn)生數(shù)據(jù)。 這個(gè)過程，相當(dāng)于throttle每往下游傳遞一個(gè)數(shù)據(jù)，都關(guān)上了上下游之間閘門，只有當(dāng)durationSelector產(chǎn)生數(shù)據(jù)的時(shí)候才打開這個(gè)閘門。到了2000毫秒的時(shí)刻，durationSelector第二次被調(diào)用產(chǎn)生的Observable對(duì)象終于產(chǎn)生了多個(gè)數(shù)據(jù)，閘門被打開，source$產(chǎn)生的第三個(gè)數(shù)據(jù)2正好趕上，被 傳遞給了下游，同時(shí)關(guān)上閘門，這時(shí)候throttle會(huì)立刻退訂上一次 durationSelector返回的Observable對(duì)象，重新將數(shù)據(jù)2作為參數(shù)調(diào)用 durationSelector來獲得一個(gè)新的Observable對(duì)象，這個(gè)新的Observable對(duì)象產(chǎn)生數(shù)據(jù)的時(shí)候，閘門才會(huì)再次打開。可見，durationSelector產(chǎn)生Observable對(duì)象只有第一個(gè)產(chǎn)生的數(shù)據(jù)會(huì)有作用，而且這個(gè)數(shù)據(jù)的產(chǎn)生時(shí)機(jī)是關(guān)鍵，至于這個(gè)數(shù)據(jù)是個(gè)什么值不重要。

=> auditTime和audit操作符 <=

說明：?這兩個(gè)都是在一個(gè)時(shí)間內(nèi)，將最后一個(gè)產(chǎn)生的值發(fā)射出去，其余的值會(huì)被忽略掉。它們之間的區(qū)別是一個(gè)使用時(shí)間范圍管理，一個(gè)使用函數(shù)管理

參數(shù)：?auditTime(時(shí)間范圍, 可選參數(shù)對(duì)象)

參數(shù)：?audit(處理函數(shù))

import { interval, auditTime } from "rxjs";
interval(1000)
 ? ?.pipe(auditTime(3000))
 ? ?.subscribe(
 ? ? ? ?(val) => console.log("auditTime:", val)
 ? ?); ? ? ? ? ?
// 第一個(gè)3s：0、1、2、3 --> 三秒末也是四秒初發(fā)出值3
// 第二個(gè)3s：4、5、6、7 --> 六秒末也是七秒初發(fā)出值7
// ...

理解：?上面的時(shí)間寫3s，所以在第一個(gè)3s內(nèi)產(chǎn)生了值0、1、2，在第3s結(jié)束的時(shí)候，產(chǎn)生了值3，根據(jù)定義，所以第一個(gè)3s發(fā)出的值是3，在物理上，第n秒結(jié)束的時(shí)候，也就是第n+1秒開始的時(shí)候，所以下一個(gè)3s是從第四秒開始，然后這個(gè)時(shí)間內(nèi)產(chǎn)生4、5、6，第7s結(jié)束的時(shí)候，產(chǎn)生值7，將其傳遞給下游...后面的值都是這樣產(chǎn)生的，也就是它發(fā)出一個(gè)值傳遞到下游之后，它會(huì)等待下一個(gè)值到達(dá)，才會(huì)開始其計(jì)時(shí)

=> sampleTime和sample操作符 <=

說明：?sampleTime的作用是搜尋一個(gè)時(shí)間范圍內(nèi)的最后一個(gè)數(shù)據(jù)，將其傳遞給下游，如果這個(gè)時(shí)間范圍里面沒有值則不會(huì)傳值到下游，然后繼續(xù)下一個(gè)時(shí)間范圍的搜尋； 而sample有點(diǎn)不同，它的參數(shù)接收一個(gè)Observable對(duì)象來控制Observable，這個(gè)參數(shù)被稱為notifier，當(dāng)notifier產(chǎn)生一個(gè)數(shù)據(jù)的時(shí)候， sample就從上游拿最后一個(gè)產(chǎn)生的數(shù)據(jù)傳給下游。

參數(shù)：?sampleTime(時(shí)間范圍, 調(diào)度器)

參數(shù)：?sample(observable對(duì)象)

interval(1000)
 ?.pipe(sampleTime(2000))
 ?.subscribe((res) => console.log("sampleTime:", res));

理解：?上面數(shù)據(jù)是每隔1s產(chǎn)生一個(gè)，然后我搜尋時(shí)間范圍是2s，第一個(gè)2s，產(chǎn)生值0、1，將1傳遞出去，繼續(xù)第二個(gè)2s的搜尋，產(chǎn)生值2、3，將3傳遞出去...以此類推

（3）去重

=> distinct操作符 <=

作用：?上游同樣的數(shù)據(jù)只有第一次產(chǎn)生時(shí)會(huì)傳給下游，其余的都被舍棄掉了，判斷是否相等使用的是===

參數(shù)：?distinct(一個(gè)函數(shù)來定制需要對(duì)比什么屬性, 一個(gè)Observable對(duì)象用于清空數(shù)據(jù))

場景一： 基本使用

import { distinct, of } from "rxjs";
of(1, 3, 2, 5, 7, 1, 2)
 ?.pipe(distinct())
 ?.subscribe((res) => console.log(res));

場景二：?對(duì)對(duì)象使用

import { distinct, of } from "rxjs";
of(
 ?{ name: "RxJS", version: "v4" },
 ?{ name: "React", version: "v15" },
 ?{ name: "React", version: "v16" },
 ?{ name: "RxJS", version: "v5" }
)
 ?// 這里規(guī)定數(shù)據(jù)中的name字段相同就算相同數(shù)據(jù)
 ?.pipe(distinct((x) => x.name))
 ?.subscribe((res) => console.log(res));

第二個(gè)參數(shù)：?distinct在運(yùn)作的時(shí)候自己會(huì)先創(chuàng)建一個(gè)集合，里面存放上游的不同數(shù)據(jù)，每次上游傳遞一個(gè)數(shù)據(jù)出來就對(duì)比集合中是否有元素跟它相等，相等就舍棄，如果上游數(shù)據(jù)無限多切都是不同的，那么這個(gè)集合就會(huì)有無限的數(shù)據(jù)在里面，這就存在數(shù)據(jù)壓力，為了解決這個(gè)問題，可以使用第二個(gè)可選參數(shù)，當(dāng)這個(gè)Observable對(duì)象產(chǎn)生數(shù)據(jù)的時(shí)候，這個(gè)集合中的數(shù)據(jù)就會(huì)被清空。

=> distinctUntilChanged操作符 <=

作用：?將上游中的連續(xù)數(shù)據(jù)過濾掉

參數(shù)：?distinctUntilChanged(比較函數(shù))

import { distinctUntilChanged, of } from "rxjs";
of(
 ?{ name: "RxJS", version: "v4" },
 ?{ name: "React", version: "v15" },
 ?{ name: "React", version: "v16" },
 ?{ name: "RxJS", version: "v5" }
)
 ?// a表示上一個(gè)值，b表示當(dāng)前值
 ?.pipe(distinctUntilChanged((a, b) => a.name === b.name))
 ?.subscribe((res) => console.log(res));

注意：?比較函數(shù)需要返回布爾值來確定由哪些屬性決定數(shù)據(jù)相等

（4）其它

=> ignoreElements操作符 <=

作用：?忽略上游所有元素，只關(guān)心complete和error事件

參數(shù)：?沒有參數(shù)

import { ignoreElements, of } from "rxjs";
of(1, 2, 3)
 ?.pipe(ignoreElements())
 ?.subscribe((res) => console.log(res));

=> elementAt操作符 <=

說明：?把上游數(shù)據(jù)當(dāng)數(shù)組，只獲取指定下標(biāo)的那?個(gè)數(shù)據(jù)，如果找不到，則拋出一個(gè)錯(cuò)誤事件，如果不想出現(xiàn)錯(cuò)誤，可以使用第二個(gè)參數(shù)，在找不到的時(shí)候，會(huì)將第二個(gè)參數(shù)做為默認(rèn)值傳遞給下游

參數(shù)：?elementAt(下標(biāo), 默認(rèn)值)

import { elementAt, of } from "rxjs";
of(1, 2, 3)
 ?.pipe(elementAt(3, "使用默認(rèn)值作為數(shù)據(jù)傳遞給下游"))
 ?.subscribe((res) => console.log(res));

=> single操作符 <=

作用：?檢查上游是否只有一個(gè)滿足對(duì)應(yīng)條件的數(shù)據(jù)，如果答案為是，就向下游傳遞這個(gè)數(shù)據(jù)；如果答案為否，就向下游傳遞一個(gè)異常

參數(shù)：?single(過濾函數(shù))

import { of, single } from "rxjs";
of(1, 2, 3)
 ?.pipe(single((x) => x % 2 === 0))
 ?.subscribe((res) => console.log(res));

七、轉(zhuǎn)化數(shù)據(jù)流

（1）映射數(shù)據(jù)

理解：?映射數(shù)據(jù)是最簡單的轉(zhuǎn)化形式。假如上游的數(shù)據(jù)是A、B、C、D的序列，那么可以認(rèn)為經(jīng)過轉(zhuǎn)化類操作符之后，就會(huì)變成f（A）、f（B）、f（C）、f（D）的序列，其中f是一個(gè)函數(shù)，作用于上游數(shù)據(jù)之后，產(chǎn)生的就是傳給下游新的數(shù)據(jù)

=> map操作符 <=

說明：?它接受一個(gè)函數(shù)作為參數(shù)，這個(gè)函數(shù)通常稱為project，指定了數(shù)據(jù)映射的邏輯，每當(dāng)上游推下來一個(gè)數(shù)據(jù)，map就把這個(gè)數(shù)據(jù)作為參數(shù)傳給map的參數(shù)函數(shù)，然后再把函數(shù)執(zhí)行的返回值推給下游

參數(shù)：?map(處理函數(shù))

import { of, map } from "rxjs";
of(1, 2, 3)
 ?.pipe(
 ? ?map((item, index) => {
 ? ? ?// 處理函數(shù)的item表示當(dāng)前值，index表示當(dāng)前值得索引
 ? ? ?console.log(item, index);
 ? ?})
 ?)
 ?.subscribe();

2）無損回壓控制

說明：?把上游在一段時(shí)間內(nèi)產(chǎn)生的數(shù)據(jù)放到一個(gè)數(shù)據(jù)集合中，當(dāng)時(shí)機(jī)合適時(shí)，把緩存的數(shù)據(jù)匯聚到一個(gè)數(shù)組或者Observable對(duì)象傳給下游，這就是無損回壓控制

=> windowTime和bufferTime操作符 <=

作用：?用一個(gè)參數(shù)來指定產(chǎn)生緩沖窗口的時(shí)間間隔，以此緩存上游的數(shù)據(jù)

參數(shù)：?windowTime(劃分區(qū)塊間隔, 內(nèi)部區(qū)塊開始間隔, 最多緩存數(shù)據(jù)個(gè)數(shù))

參數(shù)：?bufferTime(劃分區(qū)塊間隔, 內(nèi)部區(qū)塊開始間隔, 最多緩存數(shù)據(jù)個(gè)數(shù))

場景一： 基本使用

import { timer, windowTime } from "rxjs";
const source$ = timer(0, 1000);
const result$ = source$.pipe(windowTime(4000));

理解：?windowTime的參數(shù)是4000，也就會(huì)把時(shí)間劃分為連續(xù)的4000毫秒長度區(qū)塊，在每個(gè)時(shí)間區(qū)塊中，上游傳下來的數(shù)據(jù)不會(huì)直接送給下游，而是在該時(shí)間區(qū)塊的開始就新創(chuàng)建一個(gè)Observable對(duì)象推送給下游，然后在這個(gè)時(shí)間區(qū)塊內(nèi)上游產(chǎn)生的數(shù)據(jù)放到這個(gè)新創(chuàng)建的Observable對(duì)象中。在每個(gè)4000毫秒的時(shí)間區(qū)間內(nèi)，上游的每個(gè)數(shù)據(jù)都被傳送給對(duì)應(yīng)時(shí)間區(qū)間的內(nèi)部Observable對(duì)象中，當(dāng)4000毫秒時(shí)間一到，這個(gè)區(qū)間的內(nèi)部Observable對(duì)象就會(huì)完結(jié)，將結(jié)果打印出來會(huì)發(fā)現(xiàn)控制臺(tái)每隔1000毫秒打印一個(gè)數(shù)字出來，因此windowTime把上游數(shù)據(jù)傳遞出去是不需要延遲的

import { bufferTime, timer } from "rxjs";
const source$ = timer(0, 1000);
const result$ = source$
 ?.pipe(bufferTime(4000))
 ?.subscribe((res) => console.log(res));

理解：?bufferTime產(chǎn)?的是普通的Observable對(duì)象，其中的數(shù)據(jù)是數(shù)組形式， bufferTime會(huì)把時(shí)間區(qū)塊內(nèi)的數(shù)據(jù)緩存，在時(shí)間區(qū)塊結(jié)束的時(shí)候把所有緩存的數(shù)據(jù)放在一個(gè)數(shù)組再傳給下游，在控制臺(tái)你會(huì)看見每隔4秒打印一個(gè)數(shù)組，因此bufferTime把上游數(shù)據(jù)傳遞出去是需要延遲的

場景二： 第二個(gè)參數(shù)

作用：?指定每個(gè)時(shí)間區(qū)塊開始的時(shí)間間隔。

import { timer, windowTime } from "rxjs";
const source$ = timer(0, 1000);
source$.pipe(windowTime(4000, 2000)).subscribe();

理解：?windowTime使用第二個(gè)參數(shù)200之后，產(chǎn)生內(nèi)部Observable的頻率更高了，每200毫秒就會(huì)產(chǎn)生一個(gè)內(nèi)部Observable對(duì)象， 而且各內(nèi)部Observable對(duì)象中的數(shù)據(jù)會(huì)重復(fù)，例如數(shù)據(jù)2和3就同時(shí)出現(xiàn)在第一個(gè)和第二個(gè)內(nèi)部Observable對(duì)象中

import { bufferTime, timer } from "rxjs";
const source$ = timer(0, 1000);
source$
 ? ?.pipe(bufferTime(4000, 2000, 2))
 ? ?.subscribe(console.log);

理解：?對(duì)于bufferTime，因?yàn)樾枰彺嫔嫌螖?shù)據(jù)，不管參數(shù)設(shè)定的數(shù)據(jù)區(qū)間有多短，都無法預(yù)期在這段時(shí)間內(nèi)上游會(huì)產(chǎn)生多少數(shù)據(jù)，如果上游在短時(shí)間內(nèi)爆發(fā)出很多數(shù)據(jù)，那就會(huì)給bufferTime很大的內(nèi)存壓力，為了防止出現(xiàn)這種情況可以使用第三個(gè)可選參數(shù)來指定每個(gè)時(shí)間區(qū)間內(nèi)緩存的最多數(shù)據(jù)個(gè)數(shù)。

注意：?如果第一個(gè)參數(shù)比第二個(gè)參數(shù)大，那么就有可能出現(xiàn)數(shù)據(jù)重復(fù)，如果第二個(gè)參數(shù)比第一個(gè)參數(shù)大，那么就有可能出現(xiàn)上游數(shù)據(jù)的丟失。之所以說“有可能”，是因?yàn)閬G失或者重疊的時(shí)間區(qū)塊中可能上游沒有產(chǎn)生數(shù)據(jù)，所以也就不會(huì)引起上游數(shù)據(jù)的丟失和重復(fù)。從這個(gè)意義上說來，windowTime和bufferTime如果用上了第二個(gè)參數(shù)，也未必是“止損”的回壓控制。

=> windowCount和bufferCount操作符 <=

作用：?根據(jù)數(shù)據(jù)個(gè)數(shù)來決定內(nèi)部的一個(gè)Observabe需要保存多少數(shù)據(jù)。

參數(shù)：?windowCount(時(shí)間區(qū)間長度, 隔幾個(gè)數(shù)據(jù)重新開一個(gè)區(qū)間)

import { timer, windowCount } from "rxjs";
const source$ = timer(0, 1000);
source$.pipe(windowCount(4)).subscribe(console.log);

import { timer, windowCount } from "rxjs";
const source$ = timer(0, 1000);
source$.pipe(windowCount(4, 5)).subscribe(console.log);

理解：?windowCount還支持可選的第二個(gè)參數(shù)，如果不使用第二個(gè)參數(shù)，那么所有的時(shí)間區(qū)間沒有重疊部分；如果使用了第二個(gè)參數(shù)，那么第二個(gè)參數(shù)依然是時(shí)間區(qū)間的長度，但是每間隔第二個(gè)參數(shù)毫秒數(shù)，就會(huì)新開一個(gè)時(shí)間區(qū)間

說明：?對(duì)于bufferCount，和windowCount一樣，區(qū)別只是傳給下游的是緩存數(shù)據(jù)組成的數(shù)組

=> windowWhen和bufferWhen操作符 <=

說明：?它們接受一個(gè)函數(shù)作為參數(shù)，這個(gè)函數(shù)返回一個(gè)Observable對(duì)象，用于控制上游的數(shù)據(jù)分割，每當(dāng)返回的Observable對(duì)象產(chǎn)生數(shù)據(jù)或者完結(jié)時(shí)，windowWhen就認(rèn)為是一個(gè)緩沖區(qū)塊的結(jié)束，重新開啟一個(gè)緩沖窗口。bufferWhen跟這個(gè)是類似的

參數(shù)：?windowWhen(處理函數(shù))

import { timer, windowWhen } from "rxjs";
const source$ = timer(0, 100);
const closingSelector = () => {
 ?return timer(400);
};
// 被訂閱的時(shí)候windowWhen就開始?作，?先開啟?個(gè)緩沖
// 窗口，然后?刻調(diào)?closingSelector獲得?個(gè)Observable對(duì)象，
// 在這個(gè)Observable對(duì)象輸出數(shù)據(jù)的時(shí)候，當(dāng)前的緩沖窗?就關(guān)閉，
// 同時(shí)開啟?個(gè)新的緩沖窗口，然后再次調(diào)?closingSelector
// 獲得?個(gè)Observable對(duì)象
source$.pipe(windowWhen(closingSelector));

=> windowToggle和bufferToggle操作符 <=

說明：?利?Observable來控制緩沖窗口的開和關(guān)。它需要兩個(gè)參數(shù)，第一個(gè)參數(shù)是一個(gè)Observable對(duì)象，當(dāng)產(chǎn)生一個(gè)數(shù)據(jù)，代表一個(gè)緩沖窗口的開始；同時(shí)，第二個(gè)參數(shù)是一個(gè)函數(shù)，它也會(huì)被調(diào)用，用來獲得緩沖窗口結(jié)束的通知；其次函數(shù)的參數(shù)是第一個(gè)參數(shù)產(chǎn)生的數(shù)據(jù)，這樣就可以由前一個(gè)參數(shù)控制緩沖窗口的開始時(shí)機(jī)，函數(shù)控制其關(guān)閉時(shí)機(jī)，從而控制產(chǎn)生高階Observable的節(jié)奏；同理bufferToggle也是類似的

import { timer, windowToggle } from "rxjs";
const source$ = timer(0, 100);
const openings$ = timer(0, 400);
const closingSelector = (value) => {
 ?return value % 2 === 0 ? timer(200) : timer(100);
};
// opening$每400毫秒產(chǎn)??個(gè)數(shù)據(jù)，所以每400毫秒就會(huì)有?個(gè)
// 緩沖區(qū)間開始。每當(dāng)opening$產(chǎn)??個(gè)數(shù)據(jù)時(shí)，closingSelector
// 就會(huì)被調(diào)?返回控制對(duì)應(yīng)緩沖區(qū)間結(jié)束的Observable對(duì)象，
// 如果參數(shù)為偶數(shù)，就會(huì)延時(shí)200毫秒產(chǎn)??個(gè)數(shù)據(jù)，否則就延時(shí)100
// 毫秒產(chǎn)??個(gè)數(shù)據(jù)
source$.pipe(windowToggle(openings$, closingSelector));

=> window和buffer操作符 <=

說明：?保持一個(gè)Observable類型的參數(shù)，稱為notifier$，每當(dāng)notifer$產(chǎn)生一個(gè)數(shù)據(jù)，既是前一個(gè)緩存窗口的結(jié)束，也是后一個(gè)緩存窗口的開始；如果這個(gè)Observable完結(jié)了，那么window產(chǎn)生的一階Observable對(duì)象也會(huì)完結(jié)，buffer也是類似的

參數(shù)：?window(一個(gè)Observable對(duì)象)

import { timer, window } from "rxjs";
const source$ = timer(0, 100);
// 一個(gè)不會(huì)完結(jié)的Observable
const notifer$ = timer(400, 400);
source$.pipe(window(notifer$));

import { timer, window } from "rxjs";
const source$ = timer(0, 100);
// 一個(gè)會(huì)完結(jié)的Observable
const notifer$ = timer(400);
source$.pipe(window(notifer$));

（3）高階map

說明：?傳統(tǒng)map與高階map的區(qū)別在于其函數(shù)參數(shù)的返回值，前者是將一個(gè)數(shù)據(jù)映射成另一個(gè)數(shù)據(jù)，而后者是將一個(gè)數(shù)據(jù)轉(zhuǎn)變成一個(gè)Observable

import { interval, map } from "rxjs";
const source$ = interval(200);
// 這里每個(gè)數(shù)據(jù)都會(huì)轉(zhuǎn)換成一個(gè)包含數(shù)字0、1、2、3、4的
// Observable對(duì)象
source$.pipe(
 ? ? ? ? ? ?map(
 ? ? ? ? ? ? ? ?() => interval(100).take(5)
 ? ? ? ? ? ?)
 ? ? ? ?);

=> concatMap操作符 <=

說明：?可以理解成concatMap = map + concatAll

import { interval, concatMap } from "rxjs";
const source$ = interval(200);
source$.pipe(
 ? ? ? ? ? ?concatMap(
 ? ? ? ? ? ? ? ?() => interval(100).take(5)
 ? ? ? ? ? ?)
 ? ? ? ?);

理解：?第一個(gè)內(nèi)部Observable對(duì)象中的數(shù)據(jù)被完整傳遞給了 concatMap的下游，但是，第一個(gè)產(chǎn)生的內(nèi)部Observable對(duì)象沒有那么快處理，只有到第一個(gè)內(nèi)部Observable對(duì)象完結(jié)之后，concatMap才會(huì)去訂閱第二個(gè)內(nèi)部Observable，這樣就導(dǎo)致第二個(gè)內(nèi)部Observable對(duì)象中的數(shù)據(jù)排在了后面，絕不會(huì)和第一個(gè)內(nèi)部Observable對(duì)象中的數(shù)據(jù)交叉。

=> mergeMap操作符 <=

說明：?可以理解成mergeMap = map + mergeAll

注意：?一旦內(nèi)部Observable發(fā)出一個(gè)值，它就會(huì)立即將該值傳遞給下游觀察者，而不管其他內(nèi)部Observable是否已經(jīng)發(fā)出或者完成了

import { interval, mergeMap, take } from "rxjs";
const source$ = interval(200).take(2);
source$.pipe(
 ? ? ? ? ? ?mergeMap(
 ? ? ? ? ? ? ? ?() => interval(100).take(5)
 ? ? ? ? ? ?)
 ? ? ? ?);

=> switchMap操作符 <=

說明：?可以理解成switchMap = map + switchAll

注意：?后產(chǎn)生的內(nèi)部Observable對(duì)象優(yōu)先級(jí)總是更高，只要有新的內(nèi)部Observable對(duì)象產(chǎn)生，就立刻退訂之前的內(nèi)部 Observable對(duì)象，改為從最新的內(nèi)部Observable對(duì)象拿數(shù)據(jù)

import { interval, switchMap, take } from "rxjs";
const source$ = interval(200).take(2);
source$.pipe(
 ? ? ? ? ? ?switchMap(
 ? ? ? ? ? ? ? ?() => interval(100).take(5)
 ? ? ? ? ? ?)
 ? ? ? ?);

4）分組

=> groupBy操作符 <=

參數(shù)：?groupBy(一個(gè)處理函數(shù)，用于得到數(shù)據(jù)的key值)

機(jī)制：?對(duì)于上游推送下來的任何數(shù)據(jù)，檢查這個(gè)數(shù)據(jù)的key值，如果這個(gè)key值是第一次出現(xiàn)，就產(chǎn)生一個(gè)新的內(nèi)部Observable對(duì)象，同時(shí)這個(gè)數(shù)據(jù)就是內(nèi)部Observable對(duì)象的第一個(gè)數(shù)據(jù)；如果key值已經(jīng)出現(xiàn)過，就直接把這個(gè)數(shù)據(jù)塞給對(duì)應(yīng)的內(nèi)部Observable對(duì)象

import { groupBy, interval } from "rxjs";
const source$ = interval(200);
source$.pipe(groupBy((val) => val % 2));

理解：?groupBy的函數(shù)參數(shù)取的是參數(shù)除以2的余數(shù)，所以會(huì)產(chǎn)生兩個(gè)key值：0和1。從彈珠圖中可以看到，0和2屬于第一個(gè)內(nèi)部 Observable對(duì)象，第一個(gè)內(nèi)部Observable對(duì)象收納所有key值為0的數(shù)據(jù)，1 和3屬于第二個(gè)內(nèi)部Observable對(duì)象，因?yàn)樗鼈儗?duì)應(yīng)的key值為1

=> partition操作符 <=

說明：?partition接受一個(gè)判定函數(shù)作為參數(shù)，對(duì)上游的每個(gè)數(shù)據(jù)進(jìn)行判定，滿足條件的放一個(gè)Observable對(duì)象，不滿足條件的放到另一個(gè)Observable對(duì)象，就這樣來分組，它返回的是一個(gè)數(shù)組，包含兩個(gè)元素，第一個(gè)元素是容納滿組判定條件的Observable對(duì)象，第二個(gè)元素當(dāng)然是不滿足判定條件的Observable對(duì)象。

參數(shù)：?partition(數(shù)據(jù)源, 判定函數(shù))

import { partition, timer } from "rxjs";
const source$ = timer(0, 100);
// 解構(gòu)賦值
const [even$, odd$] = partition(source$, (x) => x % 2 === 0);
even$.subscribe((value) => console.log("even:", value));
odd$.subscribe((value) => console.log("odd:", value));

注意：?使用 partition一般也不會(huì)在后面直接使用鏈?zhǔn)秸{(diào)用，需要把結(jié)果以變量存儲(chǔ)，然后分別處理結(jié)果中的兩個(gè)Observable對(duì)象

（5）累計(jì)數(shù)據(jù)

=> scan操作符 <=

說明：?與reduce操作符類似，它也有一個(gè)求和函數(shù)參數(shù)和一個(gè)可選的seed種子參數(shù)作為求和初始值。scan和reduce的區(qū)別在于scan對(duì)上游每一個(gè)數(shù)據(jù)都會(huì)產(chǎn)生一個(gè)求和結(jié)果，reduce是對(duì)上游所有數(shù)據(jù)進(jìn)行求和，reduce最多只給下游傳遞一個(gè)數(shù)據(jù)，如果上游數(shù)據(jù)永不完結(jié)，那reduce也永遠(yuǎn)不會(huì)產(chǎn)生數(shù)據(jù)，scan完全可以處理一個(gè)永不完結(jié)的上游Observable對(duì)象

參數(shù)：?scan(求和函數(shù), 初始值)

import { interval, scan } from "rxjs";
const source$ = interval(1000);
source$
 ?.pipe(
 ? ?// sum：上一次求和后的值
 ? ?// current：當(dāng)前需要進(jìn)行求和的值
 ? ?scan((sum, current) => {
 ? ? ?console.log(sum, current);
 ? ? ?return sum + current;
 ? ?})
 ?)
 ?.subscribe();

理解：?scan的規(guī)約函數(shù)參數(shù)把之前求和的值加上當(dāng)前數(shù)據(jù)作為求和結(jié)果，每一次上游產(chǎn)生數(shù)據(jù)的時(shí)候，這個(gè)求和函數(shù)都會(huì)被調(diào)用，結(jié)果會(huì)傳給下游，同時(shí)結(jié)果也會(huì)由scan保存，作為下一次調(diào)用規(guī)約函數(shù)時(shí)的sum參數(shù)

=> mergeScan操作符 <=

說明：?它在使用的時(shí)候跟scan是類似的，不過它的返回值是一個(gè)Observable對(duì)象

機(jī)制：?每當(dāng)上游推送一個(gè)數(shù)據(jù)下來，mergeScan就調(diào)用一次求和函數(shù)，并且訂閱返回的Observable對(duì)象，之后，這個(gè)Observable對(duì)象會(huì)使用類似merge的方式與下游合并，此時(shí)mergeScan會(huì)記住傳給下游的最后一個(gè)數(shù)據(jù)，當(dāng)上游再次推送數(shù)據(jù)下來的時(shí)候，就把最后一次傳遞給下游的數(shù)據(jù)作為求和函數(shù)的sum參數(shù)

注意：?如果mergeScan返回一個(gè)復(fù)雜或者不會(huì)完結(jié)的Observable對(duì)象，可能會(huì)導(dǎo)致上游數(shù)據(jù)和返回的Observable對(duì)象會(huì)交叉?zhèn)鬟f數(shù)據(jù)給下游，這樣那個(gè)值是最后一次傳遞給下游的會(huì)很難確定，因此在使用的時(shí)候返回的Observable里面包含的值盡量簡單

八、錯(cuò)誤處理

說明：?錯(cuò)誤異常和數(shù)據(jù)一樣，會(huì)沿著數(shù)據(jù)流管道從上游向下游流動(dòng)，流過所有的過濾類或者轉(zhuǎn)化類操作符，最后會(huì)觸發(fā)Observer的error方法，不過也不是所有錯(cuò)誤都交給Observer處理，不然它需要處理的東西就太多了，此時(shí)就需要在數(shù)據(jù)管道中處理掉，這里處理異常有兩類方法：恢復(fù)和重試。在實(shí)際應(yīng)用中，重試和恢復(fù)往往配合使用，因?yàn)橹卦囃怯写螖?shù)限制的，不能無限重試，如果嘗試了次數(shù)上限之后得到的依然是錯(cuò)誤異常， 還是要用“恢復(fù)”的方法獲得默認(rèn)值繼續(xù)運(yùn)算。

恢復(fù)：就是本來雖然產(chǎn)生了錯(cuò)誤異常，但是依然讓運(yùn)算繼續(xù)下去。最常見的場景就是在獲取某個(gè)數(shù)據(jù)的過程中發(fā)生了錯(cuò)誤，這時(shí)候雖然沒有獲得正確數(shù)據(jù)，但是用一個(gè)默認(rèn)值當(dāng)做返回的結(jié)果，讓運(yùn)算繼續(xù)。 重試：就是當(dāng)發(fā)現(xiàn)錯(cuò)誤異常的時(shí)候，認(rèn)為這個(gè)錯(cuò)誤只是臨時(shí)的，重新嘗試之前發(fā)生錯(cuò)誤的操作，寄希望于重試之后能夠獲得正常的結(jié)果，其本質(zhì)是在訂閱上游的同時(shí)，退訂上一次訂閱的內(nèi)容

=> catchError操作符 <=

作用：?會(huì)在管道中捕獲上游傳遞過來的錯(cuò)誤

參數(shù)：?catchError(異常函數(shù))

import { range, map, catchError, of } from "rxjs";
// 產(chǎn)生數(shù)據(jù)1、2、3、4、5
const source$ = range(1, 5);
// 遍歷數(shù)據(jù)發(fā)現(xiàn)在4這個(gè)位置會(huì)拋出一個(gè)錯(cuò)誤
const error$ = source$.pipe(
 ?map((value) => {
 ? ?if (value === 4) {
 ? ? ?throw new Error("unlucky number 4");
 ? ?}
 ? ?return value;
 ?})
);
// 此時(shí)錯(cuò)誤會(huì)被catchError的處理函數(shù)所接收
const catch$ = error$
 ?.pipe(
 ? ?// err：被捕獲的錯(cuò)誤
 ? ?// caught$：上游緊鄰的那個(gè)Observable對(duì)象，此處就是指error$了
 ? ?catchError((err, caught$) => {
 ? ?
 ? ? ?// 函數(shù)的返回值是一個(gè)Observable對(duì)象，用來替代發(fā)生錯(cuò)誤的那個(gè)數(shù)據(jù)，然后傳遞給下游
 ? ? ?return of(8);
 ? ?})
 ?)
 ?// 錯(cuò)誤被catchError捕獲處理，所以此處不存在錯(cuò)誤
 ?.subscribe(console.log);

注意：?異常函數(shù)的第一個(gè)參數(shù)caught$比較有意思，因?yàn)樗淼氖巧嫌蔚?Observable對(duì)象，如果異常函數(shù)就返回caught$的話，相當(dāng)于讓上游Observable 重新試一遍，所以，catch這個(gè)操作符其實(shí)不光有恢復(fù)的功能，也有重試的功能

=> retry操作符 <=

第一種參數(shù)： 直接傳一個(gè)數(shù)字

說明：?它可以讓上游的Observable重新試一遍，以達(dá)到重試的目的，它接受一個(gè)數(shù)值參數(shù)number，number等于指定重試的次數(shù)， 如果number為負(fù)數(shù)或者沒有number參數(shù)，那么就是無限次retry，直到上游不再拋出錯(cuò)誤異常為止

參數(shù)：?retry(重試的次數(shù))

注意：?retry調(diào)用應(yīng)該有一個(gè)正整數(shù)的參數(shù)，也就是要指定有限次數(shù)的重試，否則，很可能陷入無限循環(huán)，畢竟被重試的上游Observable只是有可能重試成功，意思就是也有可能重試不成功，如果真的運(yùn)氣不好就是重試不成功，也真沒有必要一直重試下去，因?yàn)閞etry通常要限定重試次數(shù)，所以retry通常也要和catch配合使用，重試只是增加獲得成功結(jié)果的概率，當(dāng)重試依然沒有結(jié)果的時(shí)候，還是要catch上場做恢復(fù)的操作

import { range, map, catchError, of, retry } from "rxjs";
const source$ = range(1, 5);
const error$ = source$.pipe(
 ?map((value) => {
 ? ?if (value === 4) {
 ? ? ?throw new Error("unlucky number 4");
 ? ?}
 ? ?return value;
 ?})
);
const catch$ = error$
 ?.pipe(
 ? ?// 重復(fù)兩次
 ? ?retry(2),
 ? ?catchError((err, caught$) => {
 ? ? ?return of(8);
 ? ?})
 ?)
 ?.subscribe(console.log);

第二種參數(shù)： 傳一個(gè)配置對(duì)象
配置對(duì)象的取值：
count： 表示重試的次數(shù)限制。如果未指定，將會(huì)無限次重試，直到成功或者遇到無法處理的錯(cuò)誤 delay： 表示每次重試之間的延遲時(shí)間。可以是一個(gè)數(shù)字，表示固定的延遲時(shí)間，也可以是一個(gè)函數(shù)，接受錯(cuò)誤對(duì)象和重試次數(shù)作為參數(shù)，返回一個(gè) Observable 或 Promise，用于動(dòng)態(tài)計(jì)算延遲時(shí)間 resetOnSuccess： 表示是否在成功后重置重試計(jì)數(shù)。如果設(shè)置為true，則在每次成功后重置重試計(jì)數(shù)，否則會(huì)保持重試計(jì)數(shù)直到達(dá)到設(shè)定的重試次數(shù)或者遇到無法處理的錯(cuò)誤
注意：
delay地方如果寫一個(gè)函數(shù)在這里，這個(gè)函數(shù)會(huì)在發(fā)生錯(cuò)誤時(shí)被調(diào)用，它有兩個(gè)參數(shù)，一個(gè)是err$表示發(fā)生錯(cuò)誤的對(duì)象, 一個(gè)是retryCount表示當(dāng)前重試的次數(shù)，它需要一個(gè)返回值，不然函數(shù)無法正確的獲取錯(cuò)誤對(duì)象，導(dǎo)致重試不會(huì)繼續(xù)下去。
如果delay函數(shù)的返回值是一個(gè)Observable對(duì)象，那么每次這個(gè)對(duì)象吐出一個(gè)數(shù)據(jù)，就會(huì)重復(fù)一次，由此可以結(jié)合timer類似的操作符來達(dá)到延遲重復(fù)的目的

import { range, map, catchError, of, retry } from "rxjs";
const source$ = range(1, 5);
const error$ = source$.pipe(
 ?map((value) => {
 ? ?if (value === 4) {
 ? ? ?throw new Error("unlucky number 4");
 ? ?}
 ? ?return value;
 ?})
);
const catch$ = error$
 ?.pipe(
 ? ?// 重復(fù)兩次
 ? ?retry({
 ? ? ?count: 2,
 ? ? ?delay: (err$, retryCount) => {
 ? ? ? ?console.log(err$, retryCount);
 ? ? ? ?// 如果這里沒有返回值，下面只會(huì)出現(xiàn)一次重復(fù)
 ? ? ? ?return of(1000);
 ? ? ?},
 ? ?}),
 ? ?catchError((err, caught$) => {
 ? ? ?return of(8);
 ? ?})
 ?)
 ?.subscribe();

=> finalize操作符 <=

說明：?它接受一個(gè)回調(diào)函數(shù)作為參數(shù)，上游無論是完結(jié)還是出現(xiàn)錯(cuò)誤這個(gè)函數(shù)都會(huì)執(zhí)行，只不過在一個(gè)數(shù)據(jù)流中只會(huì)作用一次，同時(shí)這個(gè)函數(shù)也無法影響數(shù)據(jù)流。

九、多播

說明：?多播就是讓一個(gè)數(shù)據(jù)流的內(nèi)容被多個(gè)Observer訂閱

（1）數(shù)據(jù)流的關(guān)系

說明：?這里指的是Observable和Observer的關(guān)系，可以理解成前者播放內(nèi)容，后者接受內(nèi)容，播放的形式有單播、廣播和多播

理解概念：
單播： 就是一個(gè)播放者對(duì)應(yīng)朵個(gè)收聽者，一對(duì)朵的關(guān)系，例如，你使用微信給你的朋友發(fā)送信息，這就是單播，你發(fā)送的信息只有你的朋友才能收到
廣播： 例如，有一個(gè)好消息你不想只分享給一個(gè)人，而是想告訴所有的同事或者同學(xué)，你就在辦公室或者教室里大聲吼出這個(gè)好消息，所有人都聽見了，這就是“廣播”，不過發(fā)布消息的根本不知道聽眾是什么樣的人，于是篩選消息的責(zé)任就完全落在了接收方的人上，以至于難以控制。
多播： 假如有一些八卦消息，你想要分享給一群朋友，但并不想分享給所有人，或者不想在公共場合大聲嚷嚷，于是你在微信上把相關(guān)朋友拉進(jìn)一個(gè)群，在群里說出這個(gè)消息，只有被選中的朋友才能收到這條消息，這就叫做“多播”

（2）Subject

承上啟下：?根據(jù)第一部分對(duì)兩種Observable的理解不難得到Cold Observable實(shí)現(xiàn)的是單播，Hot Observable實(shí)現(xiàn)的多播

問題：?如何把Cold Observable變成Hot Observable呢

解決：?在函數(shù)式編程的世界里，有一個(gè)要求是保持不可變性，所以，要把一個(gè)Cold Observable對(duì)象轉(zhuǎn)換成一個(gè)Hot Observable對(duì)象，并不是去改變這個(gè)Cold Observable對(duì)象本身，而是產(chǎn)生一個(gè)新的Observable對(duì)象，包裝之前Cold Observable對(duì)象，這樣在數(shù)據(jù)流管道中，新的Observable對(duì)象就成為了下游，想要Hot數(shù)據(jù)源的Observer要訂閱的是這個(gè)作為下游的Observable對(duì)象，所以此時(shí)需要一個(gè)中間人來完成轉(zhuǎn)化，這個(gè)中間人就是Subject

中間人的職責(zé)：
要提供subscribe方法，讓其他?能夠訂閱一個(gè)的數(shù)據(jù)源，相當(dāng)于一個(gè)Observable
要能夠有辦法接受推送的數(shù)據(jù)，包括Cold Observable推送的數(shù)據(jù)，相當(dāng)于一個(gè)Observer

=> 雙重身份 <=

說明：?這里說的是它具有具Observable和Observer的性質(zhì)，雖然?個(gè)Subject對(duì)象是一個(gè)Observable，但是這兩個(gè)之間存在區(qū)別，區(qū)別在于Subject是存在記憶的，也就是它能夠記住有哪些Observer訂閱了自己，Subject有狀態(tài)，這個(gè)狀態(tài)就是所有Observer的列表，所以，當(dāng)調(diào)用Subject的next函數(shù)時(shí)，才可以把消息通知給所有的Observer

import { Subject } from "rxjs";
const subject = new Subject();
// 1號(hào)Observer訂閱了subject
const subscription1 = subject.subscribe(
 ?(value) => console.log("on observer 1 data: " + value),
 ?(err) => console.log("on observer 1 error: " + err.message),
 ?() => console.log("on observer 1 complete")
);
// 調(diào)?subject的next推送了數(shù)據(jù)1，這個(gè)消息只有1號(hào)Observer響應(yīng)，
// 因?yàn)楫?dāng)前只有?個(gè)Observer。同時(shí)因?yàn)閚ext（1）在2號(hào)Observer
// 加?之前執(zhí)?，所以2號(hào)Observer沒有接收到1
subject.next(1);
// 2號(hào)Observer也訂閱了subject
subject.subscribe(
 ?(value) => console.log("on observer 2 data: " + value),
 ?(err) => console.log("on observer 2 error: " + err.message),
 ?() => console.log("on observer 2 complete")
);
// 這時(shí)候調(diào)?subject的next?法推送數(shù)據(jù)2，subject現(xiàn)在知道??
// 有兩個(gè)Observer，所以會(huì)分別推送消息給1號(hào)和2號(hào)Observer
subject.next(2);
// subject的1號(hào)Observer通過unsubscribe?法退訂
subscription1.unsubscribe();
// 這時(shí)候subject知道??只有?個(gè)2號(hào)Observer，
// 所以，當(dāng)調(diào)?complete?法時(shí)，只有2號(hào)Observer接到通知
subject.complete();

特點(diǎn)：?后加入的觀察者，并不會(huì)獲得加入之前Subject對(duì)象上通過next推送的數(shù)據(jù)

實(shí)現(xiàn)多播：?既然Subject既有Observable又有Observer的特性，那么，可以讓一個(gè)Subject對(duì)象成為一個(gè)Cold Observable對(duì)象的下游，其他想要Hot數(shù)據(jù)源就可以訂閱這個(gè)Subject對(duì)象來達(dá)到轉(zhuǎn)換的目的，以此完成多播的操作。

=> 不能重復(fù)使用 <=

說明：?Subject對(duì)象也是一個(gè)Observable對(duì)象，但是因?yàn)樗型杲Y(jié)的狀態(tài)，所以不像Cold Observable對(duì)象一樣每次被subscribe都是一個(gè)新的開始，正因?yàn)槿绱耍琒ubject對(duì)象是不能重復(fù)使用的，所謂不能重復(fù)使用，指的是一個(gè) Subject對(duì)象一旦被調(diào)用了complete或者error函數(shù)，那么，它作為Observable 的生命周期也就結(jié)束了，后續(xù)還想調(diào)用這個(gè)Subject對(duì)象的next函數(shù)傳遞數(shù)據(jù)給下游，會(huì)沒有任何反應(yīng)。

import { Subject } from "rxjs";
const subject = new Subject();
// ?先1號(hào)Observer成為subject的下游
subject.subscribe(
 ?(value) => console.log("on observer 1 data: " + value),
 ?(err) => console.log("on observer 1 error: " + err.message),
 ?() => console.log("on observer 1 complete")
);
// 然后通過subject的next函數(shù)傳遞了1和2
subject.next(1);
subject.next(2);
// 緊接著調(diào)?了subject的complete函數(shù)，結(jié)束了subject的?命周期
subject.complete();
// 2號(hào)Observer也成為subject的下游，但是，這時(shí)候subject已經(jīng)完結(jié)了
subject.subscribe(
 ?(value) => console.log("on observer 2 data: " + value),
 ?(err) => console.log("on observer 2 error: " + err.message),
 ?() => console.log("on observer 2 complete")
);
// 后續(xù)通過next傳遞參數(shù)3的調(diào)?，不會(huì)傳遞給2號(hào)Observer，
// 也不會(huì)傳遞給1號(hào)Observer，但是可以獲取subject的complete通知，
// 可以這樣認(rèn)為，當(dāng)?個(gè)Subject對(duì)象的complete函數(shù)被調(diào)?之后，
// 它暴露給下游的Observable對(duì)象就是?個(gè)由empty變量產(chǎn)?的直接
// 完結(jié)的Observable對(duì)象
subject.next(3);

注意：?在Subject的生命周期結(jié)束之后，再次調(diào)用next方法沒有任何反應(yīng)，也不會(huì)拋出錯(cuò)誤，這樣可能會(huì)認(rèn)為上游所有數(shù)據(jù)都傳遞成功了，這是不合理的，由于Subject是一個(gè)Observable，那么它就會(huì)存在一個(gè)unsubscribe的方法，表示它已經(jīng)不管事了，再次調(diào)用其next方法就會(huì)報(bào)錯(cuò)，所以可以像下面這樣達(dá)到警示的目的。

import { Subject, interval, take } from "rxjs";
// tick$會(huì)間隔?秒鐘吐出數(shù)據(jù),調(diào)?下游subject的next函數(shù)
const tick$ = interval(1000).pipe(take(5));
const subject = new Subject();
tick$.subscribe(subject);
subject.subscribe((value) => console.log("observer: " + value));
// 在1.5秒的時(shí)候subject的unsubscribe函數(shù)被調(diào)?，
// 所以，2秒以后的時(shí)間，tick$還要調(diào)?subject的
// next就會(huì)拋出?個(gè)錯(cuò)誤異常
setTimeout(() => {
 ?subject.unsubscribe();
}, 1500);

=> 多個(gè)上游 <=

說明：?理論上可以用一個(gè)Subject合并多個(gè)Observable的數(shù)據(jù)流，但是這樣做并不合適，原因在于任何一個(gè)上游數(shù)據(jù)流的完結(jié)或者出錯(cuò)信息都可以終結(jié)Subject對(duì)象的生命。

import { Subject, interval, take, map } from "rxjs";
// 這兩個(gè)數(shù)據(jù)流都是通過interval產(chǎn)?的Cold Observable對(duì)象，
// 每隔?秒鐘吐出?個(gè)整數(shù)，然后利?map轉(zhuǎn)化為間隔?秒鐘吐出
// ?個(gè)固定的字符串，利?take只從兩個(gè)數(shù)據(jù)流中分別拿兩個(gè)數(shù)據(jù)
const tick1$ = interval(1000).pipe(
 ?map(() => "a"),
 ?take(2)
);
const tick2$ = interval(1000).pipe(
 ?map(() => "b"),
 ?take(2)
);
const subject = new Subject();
tick1$.subscribe(subject);
tick2$.subscribe(subject);
subject.subscribe((value) => console.log("observer 1: " + value));
subject.subscribe((value) => console.log("observer 2: " + value));
// tick1$每隔?秒鐘吐出?個(gè)a字符串，吐出兩個(gè)之后完結(jié)， 
// tick2$同樣每隔?秒鐘吐出?個(gè)字符串，只不過吐出的是b，
// 同樣是吐出兩個(gè)之后完結(jié)。因?yàn)閟ubject訂閱了tick1$和tick2$，
// 所以理論上結(jié)果應(yīng)該是下面這八個(gè)值，但其實(shí)并不是
// observer 1: a
// observer 2: a
// observer 1: b
// observer 2: b
// observer 1: a
// observer 2: a
// observer 1: b
// observer 2: b

理解：?為tick1$是由take產(chǎn)生的，也就是說在吐出2個(gè)數(shù)據(jù)之后就會(huì)調(diào)用下游的complete函數(shù)，也就是調(diào)用subject的complete函數(shù)，此時(shí)它已經(jīng)完結(jié)，后續(xù)的next的方法是沒有效果的，這也是為什么第二個(gè)b不會(huì)有效果的原因。

=> 錯(cuò)誤處理 <=

說明：?如果Subject有多個(gè)Observer，并且Subject的某個(gè)下游數(shù)據(jù)流產(chǎn)生了一個(gè)錯(cuò)誤異常，而且這個(gè)錯(cuò)誤異常沒有被Observer處理，那這個(gè)Subject其他的Observer都會(huì)失敗，為了避免這種情況的發(fā)生，每有一個(gè)Observer的時(shí)候，就需要給它一個(gè)處理錯(cuò)誤的地放就可以解決這個(gè)問題了。

十、調(diào)度器Scheduler

（1）作用

作用：?用于控制RxJS數(shù)據(jù)流中數(shù)據(jù)消息的推送節(jié)奏

舉例：?這里以帶Scheduler類型的參數(shù)的操作符range為例，不過使用調(diào)度器的這種寫法已經(jīng)廢棄，這里只是舉例而已

// 不使用調(diào)度器
import { range } from "rxjs";
const source$ = range(1, 3);
console.log("before subscribe");
source$.subscribe(
 ?(value) => console.log("data: ", value),
 ?(error) => console.log("error: ", error),
 ?() => console.log("complete")
);
console.log("after subscribe");

解釋：?因?yàn)閞ange是同步輸出數(shù)據(jù)，所有當(dāng)Observer添加之后，會(huì)把所有數(shù)據(jù)全部吐出，所以上面的代碼也是完全同步執(zhí)行的。

// 使用調(diào)度器，寫法已經(jīng)廢棄
import { range, asapScheduler } from "rxjs";
const source$ = range(1, 3, asapScheduler);
console.log("before subscribe");
source$.subscribe(
 ?(value) => console.log("data: ", value),
 ?(error) => console.log("error: ", error),
 ?() => console.log("complete")
);
console.log("after subscribe");

思考：?所以這里的asapScheduler決定了數(shù)據(jù)推送任務(wù)不是同步執(zhí)行，因?yàn)閞ange數(shù)據(jù)的吐出是在after subscribe字符串之后的，那么什么是Scheduler呢？

RxJS中定義Scheduler：
它是一種數(shù)據(jù)結(jié)構(gòu) 它是一個(gè)執(zhí)行環(huán)境 它有一個(gè)虛擬時(shí)鐘
解釋：
所謂Scheduer是?種數(shù)據(jù)結(jié)構(gòu)，指的是Scheduler對(duì)象可以根據(jù)優(yōu)先級(jí)或者其他某種條件來安排任務(wù)執(zhí)行隊(duì)列 Scheduler可以指定一個(gè)任務(wù)何時(shí)何地執(zhí)行，所以它是一個(gè)執(zhí)行環(huán)境在RxJS的數(shù)據(jù)流世界中，Scheduler說現(xiàn)在是幾點(diǎn)幾分幾秒，那現(xiàn)在就是幾點(diǎn)幾分幾秒，所以Scheduler就像是這個(gè)世界中的權(quán)威標(biāo)準(zhǔn)時(shí)鐘，正因?yàn)镾cheduler提供的虛擬時(shí)鐘可以被操縱，所以可以利用Scheduler來控制數(shù)據(jù)流中數(shù)據(jù)的流動(dòng)節(jié)奏。

（2）內(nèi)置的Scheduler

（3）支持Scheduler的操作符

=> observeOn操作符 <=

作用：?根據(jù)上游的Observable對(duì)象產(chǎn)生出一個(gè)新的Observable對(duì)象出來，讓這個(gè)新的Observable對(duì)象吐出的數(shù)據(jù)由指定的Scheduler來控制

參數(shù)：?observeOn(調(diào)度器)

import { range, observeOn, asapScheduler } from "rxjs";
const source$ = range(1, 3);
const asapSource$ = source$.pipe(observeOn(asapScheduler));
console.log("before subscribe");
// 訂閱新產(chǎn)生的Observable發(fā)現(xiàn)受調(diào)度器的影響
asapSource$.subscribe(
 ?(value) => console.log("data: ", value),
 ?(error) => console.log("error: ", error),
 ?() => console.log("complete")
);
console.log("after subscribe");

import { range, observeOn, asapScheduler } from "rxjs";
const source$ = range(1, 3);
const asapSource$ = source$.pipe(observeOn(asapScheduler));
console.log("before subscribe");
// 訂閱上游數(shù)據(jù)發(fā)現(xiàn)不受調(diào)度器的影響
source$.subscribe(
 ?(value) => console.log("data: ", value),
 ?(error) => console.log("error: ", error),
 ?() => console.log("complete")
);
console.log("after subscribe");

注意：?observeOn只控制新產(chǎn)生的Observable對(duì)象的數(shù)據(jù)推送節(jié)奏，并不能改變上游Observable對(duì)象所使用的Scheduler

=> subscribeOn操作符 <=

說明：?這個(gè)跟observeOn的區(qū)別在于前者是控制什么時(shí)候訂閱Observable對(duì)象，而后者是控制Observable對(duì)象何時(shí)往下游推送數(shù)據(jù)，使用和參數(shù)是類似的。