前言 這個庫在angular 中已經(jīng)集成了，所以使用起來有良好的代碼提示，但是在Vue中不行，一點提示都沒有，下面的代碼都在Vue項目中使用，以此分享自己在學(xué)習(xí)的體會：
一、初始RxJS （1）安裝與導(dǎo)入 命令
按需導(dǎo)入：
import { Observer } from "rxjs";（2）Observable的工作 說明： ?Observable可以理解成被觀察者 ，Observer就是觀察者，連接兩者的橋梁就是Observable對象的函數(shù)subscribe，同時RxJS中的數(shù)據(jù)流就是Observable對象，它實現(xiàn)了觀察者模式和迭代器模式 ，這里聊聊前者。
=> 觀察者模式 <= 解決問題： ?它需要解決在一個持續(xù)產(chǎn)生事件的系統(tǒng)中，如何分割功能，讓不同模塊只需要處理一部分邏輯。
解決方法： ?將邏輯分為發(fā)布者和觀察者，發(fā)布者只管生產(chǎn)事件，之后將事件上注冊一個觀察者，至于事件如何被觀察者處理它不關(guān)心；同樣觀察者只管將接收到的事件處理掉，不關(guān)心它是如何產(chǎn)生的。
與RxJS的聯(lián)系： ?Observable對象就是一個發(fā)布者，通過函數(shù)subscribe將其與觀察者Observer聯(lián)系起來。
import { of } from "rxjs";
// of操作符會返回一個observable對象，將傳入的內(nèi)容依次發(fā)射出來；
// 此時scoure$就是一個發(fā)布者，它產(chǎn)生的事件就是三個整數(shù)
const scoure$ = of(1, 2, 3);
// 這里console.log作為觀察者，將傳給它的內(nèi)容輸出出來，
// 不管數(shù)據(jù)是怎么產(chǎn)生的
scoure$.subscribe(console.log);處理步驟：響應(yīng)事件 ：這是觀察者的責(zé)任，也就是由subscribe的參數(shù)決定。 發(fā)布者如何關(guān)聯(lián)觀察者：也就是何時調(diào)用subscribe。 => 迭代器模式 <= 說明： ?它提供一個通用的接口來遍歷數(shù)據(jù)集合的對象，并且讓使用者不用關(guān)心這個數(shù)據(jù)集合是如何實現(xiàn)的。從數(shù)據(jù)消費的角度，迭代器實現(xiàn)分為拉和推兩種，簡單理解就是拉取數(shù)據(jù)和推送數(shù)據(jù)，RxJS屬于后者，它作為迭代器的使用者，并不需要主動去從Observable 中拉數(shù)據(jù)，而是只要subscribe上Observable對象之后，然后就能夠收到消息的推送。
=> 創(chuàng)造Observable <= 執(zhí)行過程： ?創(chuàng)建一個Observable，也就是創(chuàng)建一個發(fā)布者，這個發(fā)布者接收一個onSubscribe用于與觀察者產(chǎn)生聯(lián)系，當(dāng)發(fā)布者通過subscribe將其注冊給觀察者后，這個函數(shù)就會執(zhí)行，函數(shù)的參數(shù)就是觀察者對象，對這個對象的唯一要求就是需要存在next屬性，屬性的值是一個函數(shù)，用來接收傳過來的數(shù)據(jù)
// 0.用于定義發(fā)布者
import { Observable } from "rxjs";
// 4.觸發(fā)后這個函數(shù)的參數(shù)是觀察者的一個包裝，
// ? 它與觀察者并不等價
const onSubscribe = (observer) => {
?observer.next(1);
?observer.next (2);
?observer.next(3);
};
// 1.這里創(chuàng)建一個發(fā)布者，它存在一個onSubscribe函數(shù)與
// ? 觀察者產(chǎn)生聯(lián)系
const source$ = new Observable(onSubscribe);
// 2.創(chuàng)建一個觀察者，有一個next屬性用于接收傳過來的值
const theObserver = {
?next: (item) => console.log(item),
};
// 3.通過subscribe函數(shù) 將發(fā)布者和觀察者聯(lián)系起來，此時發(fā)
// ? 布者中的onSubscribe函數(shù)會被觸發(fā)
source$.subscribe(theObserver);
=> 延遲的Observable <= 舉例： ?如何讓上面的例子中推送每個正整數(shù)之間有一定的時間間隔？
思考： ?這個邏輯放在哪個部分更合適？
解釋： ?按照分工，發(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ù)據(jù)來處理，再不用關(guān)心數(shù)據(jù)何時產(chǎn)生。
=> 永無止境的Observable <= 說明： ?其實發(fā)布者發(fā)射的數(shù)據(jù)可以是無窮的，每次發(fā)布者使用next發(fā)射出一個數(shù)據(jù)，這個數(shù)據(jù)會被觀察者接收然后消化掉，所以不會存在數(shù)據(jù)堆積；如果發(fā)布者的next方法停止調(diào)用，只能表示發(fā)布者此時不會發(fā)射數(shù)據(jù)出去，但并不代表之后不會發(fā)射數(shù)據(jù)；如果需要明確發(fā)布就不會再有新數(shù)據(jù)產(chǎn)生了，還需要多個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);
=> 錯誤的Observable <= 說明： ?理想情況下，發(fā)布者只管生產(chǎn)數(shù)據(jù)給觀察者來消耗，但是，難免有時候發(fā)布者會遇到了異常情況，而且這種異常情況不是生產(chǎn)者所能夠處理并恢復(fù)正常的，發(fā)布者在這時候沒法再正常工作了，就需要通知對應(yīng)的觀察者發(fā)生了這個異常情況，如果只是簡單地調(diào)用 complete，觀察者只會知道沒有更多數(shù)據(jù)，卻不知道沒有更多數(shù)據(jù)的原因是因為遭遇了異常，所以，我們還要在發(fā)布者和觀察者的交流渠道中增加一個新的函數(shù)error。
import { Observable } from "rxjs/Observable";
const onSubscribe = (observer) => {
?observer.next(1);
?// 此時發(fā)布者出現(xiàn)不能自己解決的錯誤，調(diào)用方法通知觀察者，
?// 此時發(fā)布者已經(jīng)進(jìn)入完結(jié)的狀態(tài)，后面所調(diào)用的next和complete
?// 都會失效
?observer.error("Someting Wrong");
?observer.complete();
};
const source$ = new Observable(onSubscribe);
const theObserver = {
?next: (item) => console.log(item),
?// 用來處理錯誤信息
?error: (err) => console.log(err),
?complete: () => console.log("No More Data"),
};
source$.subscribe(theObserver);在RxJS中，一個發(fā)布者對象只有一種終結(jié)狀態(tài)，要么是complete，要么是error，一旦進(jìn)入出錯狀態(tài)，這個發(fā)布者對象也就終結(jié)了，再不會調(diào)用對應(yīng)觀察者的next函數(shù) ，也不會再調(diào)用觀察者的complete函數(shù) ；同樣，如果一個發(fā)布者對象進(jìn)入了完結(jié)狀態(tài)，也不能再調(diào)用觀察者的next和error。 （3）退訂Observable 說明： ?有時候需要斷開發(fā)布者與觀察者之間的聯(lián)系，這個操作就叫做退訂，在發(fā)布者的onSubscribe函數(shù)執(zhí)行的時候，它可以返回一個對象，對象上可以有一個unsubscribe函數(shù)，執(zhí)行這個函數(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ù)，因為發(fā)布者并沒有到達(dá)終結(jié)狀態(tài) ，也就是沒有調(diào)用complete 或者是error 方法，此時只是發(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: () => {
? ? ?// 這里不清除定時器，讓發(fā)布者繼續(xù)產(chǎn)生數(shù)據(jù)
? ? ?// clearInterval(handle);
? ?},
?};
};
const source$ = new Observable(onSubscribe);
// 每次觀察者執(zhí)行的時候打印出收到的數(shù)據(jù)
const subscription = source$.subscribe((item) => console.log(item));
setTimeout(() => {
?subscription.unsubscribe();
}, 3500);發(fā)布者產(chǎn)生的事件，只有觀察者通過訂閱之后才會收到，在退訂之后就不會收到。 （4）了解兩種Observable 說明： ?這里介紹的是Hot Observable和Cold Observable。
場景： ?假設(shè)每個發(fā)布者對象有兩個觀察者對象來訂閱， 而且這兩個觀察者對象并不是同時訂閱，第一個觀察者對象訂閱N秒鐘之后，第二個觀察者對象才訂閱同一個發(fā)布者對象，而且，在這N秒鐘之內(nèi)，發(fā)布者對象已經(jīng)吐出了一些數(shù)據(jù)，此時對這吐出的數(shù)據(jù)有兩種處理：
Hot Observable：只需要接受從訂閱那一刻開始發(fā)布者產(chǎn)生的數(shù)據(jù)就行；有點類似在電視上面看節(jié)目，你所看到的內(nèi)容是節(jié)目當(dāng)前這一刻開始的，之前的節(jié)目你是看不見的，假如你的家人跟你一起看，那么你們看到的節(jié)目是一樣的，這就可以理解為獲取數(shù)據(jù)的數(shù)據(jù)源是相同的 Cold Observable：不能錯過，需要獲取發(fā)布者之前產(chǎn)生的數(shù)據(jù)，也就是每次都需要獲取發(fā)布者完整的數(shù)據(jù)，可以理解為每次得到的數(shù)據(jù)與之前的數(shù)據(jù)之間并不存在聯(lián)系，是相互獨立的，也就是每次會得到獨立的數(shù)據(jù)源，就像你在手機(jī)應(yīng)用市場下載游戲，跟你在同樣地方下載的游戲是一樣的。 理解： ?那么就可以得到這樣的結(jié)果，如果Cold Observable沒有訂閱者，數(shù)據(jù)不會真正的產(chǎn)生，就像你如果不主動下載游戲，你手機(jī)上不可能玩到的；而對于Hot Observable在沒有訂閱者的時候，數(shù)據(jù)依然產(chǎn)生，只不過不傳入數(shù)據(jù)管道而已，就像電視機(jī)節(jié)目，節(jié)目一直存在與此，只是你沒切換到那個頻道觀看而已。
（5）操作符簡介 說明： ?一個發(fā)布者對象就是一個數(shù)據(jù)流，在RxJS中數(shù)據(jù)流一般使用$開頭來命名；在一個復(fù)雜問題里面，數(shù)據(jù)流并不會直接交給觀察者來處理，在這途中會使用一系列內(nèi)置的函數(shù)來處理數(shù)據(jù)，這些函數(shù)可以理解為操作符；就像一個管道，數(shù)據(jù)從管道的一段流入，途徑管道各個環(huán)節(jié)，當(dāng)數(shù)據(jù)到達(dá)觀察者的時候，已經(jīng)被管道操作過，有的數(shù)據(jù)已經(jīng)被中途過濾拋棄掉了，有的數(shù)據(jù)已經(jīng)被改變了原來的形態(tài)，而且最后的數(shù)據(jù)可能來自多個數(shù)據(jù)源，最后觀察者只需要處理能夠達(dá)到終點的數(shù)據(jù)。
說明： ?在數(shù)據(jù)管道中流淌的數(shù)據(jù)就像是水，從上游流向下游。對一個操作符來說，上游可能是一個數(shù)據(jù)源，也可能是其他操作符，下游可能是最終的觀察者，也可能是另一個操作符，每一個操作符之間都是獨立的，正因為如此，所以可以對操作符進(jìn)行任意組合，從而產(chǎn)生各種功能的數(shù)據(jù)管道。
6）理解彈珠圖 作用： ?RxJS中每一個發(fā)布者是一個數(shù)據(jù)流，簡單的數(shù)據(jù)流可以由大腦想象出來，但是復(fù)雜的可就不好像了，此時就可以使用彈珠圖 來具體的方式來描述數(shù)據(jù)流，看兩張圖：
說明： ?這個彈珠圖所表示的數(shù)據(jù)流，每間隔一段時間吐出一個遞增的正整數(shù)，吐出到3的時候結(jié)束。因為每一個吐出來的數(shù)據(jù)都像是一個彈珠，所以這種表達(dá)方式叫做彈珠圖。在彈珠圖中，每個彈珠之間的間隔，代表的是吐出數(shù)據(jù)之間的時間間隔，通過這種形式，能夠很形象地看清楚每個發(fā)布者對象中數(shù)據(jù)的分布。 根據(jù)彈珠圖的傳統(tǒng)，豎杠符號|代表的是數(shù)據(jù)流的完結(jié)，對應(yīng)調(diào)用complete函數(shù)，數(shù)據(jù)流吐出數(shù)據(jù)3之后立刻就完結(jié)了。 符號×代表數(shù)據(jù)流中的異常，對應(yīng)于調(diào)用下游的error函數(shù)。
注意： ?為了描述操作符的功能，彈珠圖中往往會出現(xiàn)多條時間軸，因為各部分操作符的工作都是把上游的數(shù)據(jù)轉(zhuǎn)為傳給下游的數(shù)據(jù)，在彈珠圖上必須把上下游的數(shù)據(jù)流都展現(xiàn)出來，此外，編寫彈珠圖可以去此處，后面如果存在彈珠圖的地方所使用的代碼復(fù)制到此處就可以看到了。
二、實現(xiàn)操作符 理解： ?一個操作符是返回一個Observable對象的函數(shù)，不過，有的操作符是根據(jù)其他Observable對象產(chǎn)生返回的Observable對象，有的操作符則是利用其他類型輸出產(chǎn)生返回的Observable對象，還有一些操作符不需要輸出就可以憑空創(chuàng)造一個Observable對象，這里以實現(xiàn)一個操作符來慢慢理解什么是操作符。
（1）實現(xiàn)操作符函數(shù) 說明： ?每一個操作符是一個函數(shù)，不管函數(shù)的功能是怎樣的，它需要包含以下功能點，這里實現(xiàn)map操作符為例
返回?個全新的Observable對象。 需要存在訂閱和退訂的操作。 處理異常情況。 及時釋放資源。 => 返回Observable對象 <= 分析： ?首先map操作符的功能是遍歷得到的數(shù)據(jù)，通過傳入的參數(shù)函數(shù)來處理這些數(shù)據(jù)，看下面的例子：
// 這里的函數(shù)參數(shù)將數(shù)據(jù)的每一個值都乘以2，
// 如果source$是?個 1、2、3的序列，
// 那么map返回的序列就是2、4、6，根據(jù)函數(shù)式編程 的原則，
// map函數(shù)是不會修改原始的數(shù)據(jù)的，同時其返回值是?個全
// 新的Observable對象，這樣可以保持原始Observable對象的狀態(tài)
// 避免不可預(yù)料的行為
const result$ = source$.map(x => x * 2);實現(xiàn)： ?根據(jù)上面的分析可以得到下面這個函數(shù)
// 這里的project就是傳遞給map操作符的函數(shù)參數(shù)
function map(project) {
?// map中利?new關(guān)鍵字創(chuàng)造了?個Observable對象，
?// 函數(shù)返回的結(jié)果就是這個對象，如此?來，
?// map可以?于鏈?zhǔn)秸{(diào)?，可以在后?調(diào)?其他的操作符，
?// 或者調(diào)?subscribe增加Observer。
?return new Observable((observer) => {
? ?// 假設(shè)此處this表示發(fā)布者對象，訂閱后數(shù)據(jù)就會交給觀察者了
? ?this.subscribe({
? ? ?next: (value) => observer.next(project(value)),
? ? ?error: (err) => observer.error(error),
? ? ?complete: () => observer.complete(),
? ?});
?});
}=> 退訂處理 <= 說明： ?作為一個通用的操作符，無法預(yù)料上游Observable是如何實現(xiàn)的，上游完全可能在被訂閱時分配了特殊資源，如果不明確地告訴上游這些資源再也用不著了的話，它也不會釋放這些資源，此時就會造成資源的泄露，所以下游退訂那些資源，就要告訴上游退訂那些資源。
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ù)前面的了解這里需要存在一個unsubscribe
? ? ?// 方法用于退訂
? ? ?unsubscribe: () => {
? ? ? ?sub.unsubscribe();
? ? ?},
? ?};
?});
}=> 處理異常 <= 說明： ?上面代碼中的參數(shù)project可以輸入的情況有很多，可能存在執(zhí)行的時候不合理的代碼，此時就會出現(xiàn)異常，此時需要 捕獲異常錯誤，把異常錯誤沿著數(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 使用原型： ?這個操作符在使用的時候需要一個Observable對象實例，因此這個操作符是一個實例操作符，此時使用打補(bǔ)丁的方式關(guān)聯(lián)發(fā)布者對象的格式為Observable.prototype.操作符 = 操作符函數(shù)，既然有實例操作符，當(dāng)然也有靜態(tài)操作符，它不需要Observable實例就可以使用，它的打補(bǔ)丁的格式為Observable.操作符 = 操作符函數(shù)，這個的弊端在于會影響每一個Observable對象。
Observable.prototype.map = map;使用call和bind： ?解決上面的問題，可以讓我們?定義的操作符只對指定的 Observable對象可?，這時就可以?bind ，當(dāng)然也可以使用call 。
// 一般使用
const result$ = map.bind(Observable對象)(x => x * 2);
// 鏈?zhǔn)秸{(diào)用
const result$ = map.bind(
? ? ? ? ? ? ? ? ? ?map.bind(Observable對象)((x) => x * 2)
? ? ? ? ? ? ? ?)((x) => x + 1);
// 一般使用
onst result$ = map.call(Observable對象, x => x * 2);
// 鏈?zhǔn)秸{(diào)用
const result$ = map.call(
? ?map.call(Observable對象, (x) => x * 2),
? ?(x) => x * 2
);3）改進(jìn)操作符 說明： ?如果遵循函數(shù)式編程思想，需要使用純函數(shù)，也就是函數(shù)執(zhí)行的結(jié)果完全由輸入的參數(shù)決定，但是上面定義的函數(shù)中存在this，這是一個不確定的因素，也就是這個函數(shù)不屬于純函數(shù)了，所以在此處需要改進(jìn)一下。
=> 缺陷 <= 說明： ?在現(xiàn)代網(wǎng)頁開發(fā)的過程中，都會經(jīng)過打包才發(fā)布到產(chǎn)品環(huán)境，在打包的過程中會使用Tree-Shaking這個工具來去除代碼中沒有使用的代碼，比如那些引入的變量但是并沒有使用這種的；但是這個工具對于RxJS來說沒什么用，這是因為Tree Shaking只能做靜態(tài)代碼檢查，并不是在程序運行時去檢測這個函數(shù)是否被真的調(diào)用，只有這個函數(shù)在任何代碼中間都沒有引用過，才認(rèn)為這個函數(shù)不會被引用。然而，RxJS中任何一個操作符都是掛在 Observable類上或者Observable.prototype上的，賦值給Observable或者 Observable.prototype上某個屬性在Tree Shaking看來就是被引用，所以，所有的操作符，不管真實運用時是否被調(diào)用，都會被Tree Shaking認(rèn)為是會用到的代碼，也就不會當(dāng)做死代碼刪除；其次上面關(guān)聯(lián)Observable的方式是直接添加到其原型上面，由于全局存在一個Observable對象，就跟window對象一樣，像上面添加屬性和方法是不可取的，是會帶來隱患的。
=> 不"打補(bǔ)丁" <= 說明： ?開發(fā)RxJS庫的規(guī)則的其中一條就是不能使用打補(bǔ)丁的方式使操作符函數(shù)與Observable對象關(guān)聯(lián)起來。如果是實例操作符，可以使用前面介紹過的bind/call，讓一個操作符函數(shù)只對一個具體的Observable對象生效；如果是靜態(tài)操作符，直接使用就好。
// 這里使用上面實現(xiàn)的map函數(shù)實現(xiàn)一個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對象作為this值傳遞給map函數(shù)
? ?return map.call(this, (x) => x * 2);
};
// of操作符用于創(chuàng)建一個Observable對象
const source$ = of(1, 2, 3);
const result$ = source$.double().subscribe((res) => console.log(res));（4）lettable/pipeable操作符 原因： ?上面使用call/bind方法在函數(shù)體內(nèi)還是會使用this，函數(shù)還是不純，其次call的返回類型是無法確定的，在ts中只能使用any表示，因此會讓其失去類型檢查。
說明： ?從RxJS v5.5.0開始，加上了這種更先進(jìn)的操作符定義和使用方式，稱為pipeable操作符，也曾經(jīng)被稱為lettable操作符，但是因為字面上太難理解，所以改成pipeable。
=> let操作符 <= 作用： ?實際上就是把上游的Observable對象作為參數(shù)傳遞給let操作符里面的參數(shù)進(jìn)行處理，處理完之后將返回的Observable交給下游來訂閱。
// 下面的map函數(shù)就是上面寫的那個，這是以前的寫法，現(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$是?個純函數(shù)。
const double$ = obs$ => obs$.map(x => x * 2);
// let的參數(shù)是?個函數(shù)，在這個例?中函數(shù)參數(shù)名為double$，
// 這個函數(shù)名也以$為后綴，代表它返回的是?個Observable對象，
// double$同樣接受?個Observable對象作為輸?參數(shù)，也就是說，
// double$的功能就是根據(jù)?個Observable對象產(chǎn)??個新的
// Observable對象。
const result$ = source$.let(double$);過程： ?let起到連接上游下游 的作用，真正的工作完全由函數(shù)參數(shù)map來執(zhí)行。
處理之前的map函數(shù)： ?此時map的實現(xiàn)部分也看不到對this的訪問，而是用一個參數(shù)obs$代替了 this，這樣，在數(shù)據(jù)管道中上游的Observable是以參數(shù)形式傳遞，而不是靠 this來獲得，讓map徹底成了一個純函數(shù)。map執(zhí)行返回的結(jié)果是一個函數(shù)，接受一個Observable對象返回一個 Observable 對象，正好滿足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(),
? ?})
?);好處： ?由于每一個lettable操作符都是純函數(shù)，而且也不會被作為補(bǔ)丁掛在Observable上，Tree Shaking就能夠找到根本不會被使用的操作符。
=> pipe操作符 <= 原因： ?要導(dǎo)入let這個操作符，又不得不用傳統(tǒng)的打補(bǔ)丁或者使用call的方式，使用起來要導(dǎo)入很麻煩；所以創(chuàng)建了pipe操作符，它可以滿足let具備的功能。使用pipe只需像使用let那樣導(dǎo)入模塊，任何Observable對象都保持pipe，此外還有管道功能，可以把多個lettable操作符串接起來，形成數(shù)據(jù)管道。
import { map, filter, of } from "rxjs";
const source$ = of(1, 2, 3);
// 可以一次使用多個操作符
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對象，這些操作符可以憑空或者根據(jù)其他數(shù)據(jù)源創(chuàng)造出?個Observable對象，其次創(chuàng)建類操作符往往不會從其他Observable對象獲取數(shù)據(jù)，因為在數(shù)據(jù)管道中它自己就是數(shù)據(jù)流的源頭，基于這些特性大部分的創(chuàng)建類操作符都是靜態(tài)操作符。
（2）創(chuàng)建同步數(shù)據(jù)流 說明： ?對于同步的Observable對象，需要關(guān)心的是存在哪些數(shù)據(jù)和數(shù)據(jù)之間的先后順序，由于數(shù)據(jù)之間的時間間隔不存在因此不需要考慮時間方面的問題。
=> of操作符 <= 作用： ?可以輕松創(chuàng)建指定數(shù)據(jù)集合的Observable對象；
參數(shù)： ?of(數(shù)據(jù)1,數(shù)據(jù)2,數(shù)據(jù)3...)；
注意： ?of操作符所產(chǎn)生的Observable對象被訂閱后會將參數(shù)依次吐出來，吐完之后會調(diào)用complete方法；吐的這個過程是同步的，也就是所有數(shù)據(jù)之間是不存在間隔的。
const { of } = Rx;
of(1).pipe();const { of } = Rx;
of(1, 2, 3).pipe();
值： ?of產(chǎn)生的是Cold Observable，對于每一個Observer都會重復(fù)吐出同樣的一組數(shù)據(jù)，所以可以反復(fù)使用。
=> range操作符 <= 作用： ?對需要產(chǎn)生多個很長連續(xù)數(shù)字序列的場景，就是得上range這個操作符了，range的含義就是“范圍”，只需要指定一個范圍的開始值和長度，range 就能夠產(chǎn)生這個范圍內(nèi)的依次+1的數(shù)字序列；同樣數(shù)據(jù)吐完之后會調(diào)用complete方法。
參數(shù)： ?range(序列開始的任意數(shù)字,序列的長度)
const { range } = Rx;
range(1, 100).pipe();局限性： ?無法規(guī)定每次遞增的大小
=> generate操作符 <= 作用： ?類似一個for循環(huán)，設(shè)定一個初始值，每次遞增這個值，直到滿足某個條件的時候才中止循環(huán)，同時，循環(huán)體內(nèi)可以根據(jù)當(dāng)前值產(chǎn)生數(shù)據(jù)。
參數(shù)： ?generate(初始值, 條件判斷函數(shù), 值如何增加函數(shù), 返回結(jié)果處理函數(shù))
// 假設(shè)存在這樣的for循環(huán)：產(chǎn)??個?10?的所有偶數(shù)的平?
const result = [];
for (let i = 2; i < 10; i += 2) {
?result.push(i * i);
}
// 使用generate類似實現(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ù)參數(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ù)的時間間隔})
import { of, repeat } from "rxjs";
of(1, 2, 3)
?.pipe(
? ?repeat({
? ? ?count: 10,
? ? ?delay: 1000,
? ?})
?)
?.subscribe((res) => console.log(res));注意： ?保證上游Observable對象一定會完結(jié)。
=> EMPTY常量 <= 作用： ?產(chǎn)生一個直接完結(jié) 的Observable對象，沒有參數(shù)，不產(chǎn)生任何數(shù)據(jù)，直接完結(jié)。
const { EMPTY ?} = Rx;
EMPTY.pipe()
=> throwError操作符 <= 作用： ?它所產(chǎn)生的Observable對象也是什么都不做，直接出錯 ，拋出的錯誤就是throw的參數(shù)
參數(shù)： ?throwError(錯誤程序)
const { throwError ?} = Rx;
throwError(new Error('這是一個錯誤')).pipe()
=> NEVER常量 <= 作用： ?產(chǎn)生的Observable對象就真的是什么都不做，既不吐出數(shù)據(jù)，也不完結(jié) ，也不產(chǎn)生錯誤 ，就這樣待著，一直到永遠(yuǎn)
const { NEVER ?} = Rx;
NEVER.pipe()
（3）創(chuàng)建異步數(shù)據(jù)流 說明： ?就是創(chuàng)建異步的Observable對象，不光要考慮產(chǎn)生什么數(shù)據(jù)，還需要考慮數(shù)據(jù)之間的時間間隔了
=> interval操作符 <= 作用： ?定時從Observable對象吐出一個數(shù)據(jù)，如果不主動結(jié)束就一直執(zhí)行
參數(shù)： ?interval(吐數(shù)據(jù)的間隔毫秒數(shù))
const { interval } = Rx;
interval(1000).pipe()注意： => timer操作符 <= 作用： ?產(chǎn)生的Observable對象在指定毫秒之后會吐出一個數(shù)據(jù)，執(zhí)行后立即結(jié)束
參數(shù)： ?timer(毫秒數(shù) / 一個Date對象, 時間間隔)
// 明確延時產(chǎn)?數(shù)據(jù)的時間間隔
const { timer } = Rx;
timer(1000).pipe()// 明確的是?個時間點
const { timer } = Rx;
timer(
?new Date(
? ?new Date().getTime() + 1000
?)
).pipe()const { timer } = Rx;
timer(1000,2000).pipe()注意： => from操作符 <= 作用： ?以把任何可迭代對象都轉(zhuǎn)化為Observable對象
參數(shù)： ?from(任何可迭代對象)
const { from } = Rx;
from([1,2,3]).pipe()const { from } = Rx;
from('abc').pipe()注意： ?在from的眼中，把輸出參數(shù)都當(dāng)做一個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對象就會吐出Promise成功的結(jié)果，并且立刻結(jié)束，如果以失敗而告終的時候，from產(chǎn)生的Observable對象也會立刻產(chǎn)生失敗事件
=> fromEvent操作符 <= 作用1： ?在網(wǎng)頁開發(fā)中，可以把DOM中的事件轉(zhuǎn)化為Observable對象中的數(shù)據(jù)
參數(shù)1： ?fromEvent(事件源, 事件名稱)
// 希望點擊id為clickMe的按鈕時，id為text的div中的數(shù)字會增加1，
// 連續(xù)點擊那個按鈕，對應(yīng)數(shù)字會持續(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é)點
// 這里展示從Node.js的events中獲得數(shù)據(jù)
import { fromEvent } from "rxjs";
// 這個模塊需要使用 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ā)送任何名稱的事件，
// 第?個參數(shù)就是事件名稱，第?個參數(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ù)都會立刻吐出來，等不到訂閱的時候，當(dāng)添加了Observer的時候，仍然什么數(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對象被訂閱 和退訂 時的動作
參數(shù)： ?fromEventPattern(被訂閱時觸發(fā)的函數(shù), 被退訂時觸發(fā)的函數(shù))
import { fromEventPattern } from "rxjs";
import EventEmitter from "events";
const emitter = new EventEmitter();
// handler參數(shù)可以理解為觀察者對象中的next方法
const addHandler = (handler) => {
?// 監(jiān)聽事件源中的msg事件，每次觸發(fā)事件執(zhí)行next方法
?emitter.addListener("msg", handler);
};
const removeHandler = (handler) => {
?// 與上面相反，會移除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)聽的事件被取消掉，
// 所以此處的值并不會出現(xiàn)在Observable對象里面
subscription.unsubscribe();
emitter.emit("msg", "end");說明： ?它提供的就是一種模式，不管數(shù)據(jù)源是怎樣的行為，最后的產(chǎn)出都是一個Observable對象
=> ajax操作符 <= 作用： ?用于發(fā)送請求并根據(jù)結(jié)果返回Observable對象
參數(shù)： ?ajax('請求的地址')
// 根據(jù)github上的api獲取RxJS項?獲得的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(一個函數(shù)，這個函數(shù)會在被訂閱時調(diào)用)
// 延遲發(fā)送請求
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）合并類操作符 說明： ?其作用在于將有多個Observable對象作為數(shù)據(jù)來源，把不同來源的數(shù)據(jù)根據(jù)不同的規(guī)則合并到一個Observable對象中。
=> concat操作符 <= 作用： ?把多個Observable中的數(shù)據(jù)內(nèi)容依次合并，合并的時候原有數(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)
? ?);注意： ?它在工作的時候，會先從第一個Observable對象中獲取數(shù)據(jù)，等它complete之后，再從下一個對象中去數(shù)據(jù)，直到取完所有的，此時，如果其中有一個對象是不完結(jié)的狀態(tài)，那么它之后的Observable對象就不會有被取到的機(jī)會。
=> merge操作符 <= 作用： ?一定性訂閱上游所有的Observable對象，只要有數(shù)據(jù)傳遞下來，這個數(shù)據(jù)就會被傳遞給下游，也就是數(shù)據(jù)采取先到先出的原則，同時合并的時候原有數(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ù)一個A，重復(fù)的次數(shù)為5次
const source1$ = of("A").pipe(
? ? ? ? ? ? ? ? ? ? ? ?repeat(
? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 700 }
? ? ? ? ? ? ? ? ? ? ? ?));
// 隔800ms重復(fù)一個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ù)字，表示可以同時合并的個數(shù)
import { merge, of, repeat, pipe } from "rxjs";
// 隔700ms重復(fù)一個A，重復(fù)的次數(shù)為5次
const source1$ = of("A").pipe(
? ? ? ? ? ? ? ? ? ? ? ?repeat(
? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 700 }
? ? ? ? ? ? ? ? ? ? ? ?));
// 隔800ms重復(fù)一個B，重復(fù)的次數(shù)為5次
const source2$ = of("B").pipe(
? ? ? ? ? ? ? ? ? ? ? ?repeat(
? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 800 }
? ? ? ? ? ? ? ? ? ? ? ?));
? ? ? ? ? ? ? ? ? ? ? ?
// 隔900ms重復(fù)一個C，重復(fù)的次數(shù)為5次
const source3$ = of("C").pipe(
? ? ? ? ? ? ? ? ? ? ? ?repeat(
? ? ? ? ? ? ? ? ? ? ? ? ? ?{ count: 5, delay: 900 }
? ? ? ? ? ? ? ? ? ? ? ?));
? ? ? ? ? ? ? ? ? ? ? ?
// 限定合并的個數(shù)為2
const merged$ = merge(source1$, source2$, source3$, 2);
merged$.subscribe((res) => console.log(res));場景三：合并多個事件 說明： ?一個元素存在click 事件和touch 事件，對應(yīng)網(wǎng)頁和移動設(shè)備，假如其事件處理程序的邏輯一致，此時就可以分別使用fromEvent 獲取單個事件流，之后用merge 合并成一個數(shù)據(jù)流，就可以集中管理了
// 可以像這樣處理
const click$ = fromEvent(element, 'click');
const touchend$ = fromEvent(element, 'touchend');
merge(click$, touchend$).subscribe(處理函數(shù));=> zip操作符 <= 作用： ?將數(shù)據(jù)流里面的數(shù)據(jù)一一對應(yīng) 并使用數(shù)組 組合起來
參數(shù)： ?zip(數(shù)據(jù)流1, 數(shù)據(jù)流2, 數(shù)據(jù)流3...)
場景一： 一對一合并 import { interval, of, zip } from "rxjs";
// 一個異步數(shù)據(jù)流，產(chǎn)生的數(shù)據(jù)是無限的
const source1$ = interval(1000);
// 一個同步數(shù)據(jù)流，產(chǎn)生的數(shù)據(jù)流是有限
const source2$ = of("a", "b", "c");
// 將兩個數(shù)據(jù)流合并
zip(source1$, source2$)
? ?.subscribe(
? ? ? ?(res) => console.log(res),
? ? ? ?null,
? ? ? ?() => console.log('complete')
? ?);注意： ?這里數(shù)據(jù)的匹配是一一對應(yīng) 的，所以數(shù)據(jù)少的那個Observable決定zip 產(chǎn)生數(shù)據(jù)的個數(shù)；然后在對應(yīng)的時候需要雙方都有數(shù)據(jù) 才能夠?qū)?yīng)，這也是為什么上面的打印會隔1s才打印。
問題： 數(shù)據(jù)積壓 說明： ?如果某個上游source1$吐出數(shù)據(jù)的速度很快，而另一個上游source2$吐出數(shù)據(jù)的速度很慢，那zip就不得不先存儲source1$吐出的數(shù)據(jù)，因為RxJS的工作方式是“推”， Observable把數(shù)據(jù)推給下游之后就沒有責(zé)任保存數(shù)據(jù)了。被source1$推送了數(shù)據(jù)之后，zip就有責(zé)任保存這些數(shù)據(jù)，等著和source2$未來吐出的數(shù)據(jù)配對。假如source2$遲遲不吐出數(shù)據(jù)，那么zip就會一直保存source1$沒有配對的數(shù)據(jù)，然而這時候source1$可能會持續(xù)地產(chǎn)生數(shù)據(jù)，最后zip積壓的數(shù)據(jù)就會越來越多，占用的內(nèi)存也就越來越多。對于數(shù)據(jù)量比較小的Observable對象，這樣的數(shù)據(jù)積壓還可以忍受，但是對于超大量的數(shù)據(jù)流，使用zip就不得不考慮潛在的內(nèi)存壓力問題。
=> combineLatest操作符 <= 作用： ?合并上游所有Observable一個最新的數(shù)據(jù)，也就是它返回值是一個數(shù)組
參數(shù)： ?combineLatest([數(shù)據(jù)1, 數(shù)據(jù)2, 數(shù)據(jù)3 ...], 處理函數(shù))
場景一： 基本使用 import { combineLatest, timer } from "rxjs";
// 隔1s產(chǎn)生一個數(shù)字
const firstTimer = timer(1000, 1000);
// 隔2s產(chǎn)生一個數(shù)字
const secondTimer = timer(1000, 2000);
// 合并上面的數(shù)據(jù)流
const combinedTimers = combineLatest([firstTimer, secondTimer]);
combinedTimers.subscribe((value) => console.log(value));注意： ?首先還是一一對應(yīng)的關(guān)系，也就是如果一個數(shù)據(jù)源還沒發(fā)射值出來，那么會等待它將值發(fā)射出來，如果值沒有改變并且操作沒有完結(jié)的話，發(fā)射的值將一直是這一個，所以只有所有的Observable對象完結(jié)，combineLatest才會給下游一個complete信號，表示不會有任何數(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在工作的時候，會按照順序依次訂閱所有上游的Observable對象，只有所有上游Observable對象都已經(jīng)吐出數(shù)據(jù)了，才會給下游傳遞所有上游“最新數(shù)據(jù)”組合的數(shù)據(jù)
解釋： ?由于of產(chǎn)生的同步數(shù)據(jù)流，在被訂閱時就會吐出所有數(shù)據(jù)，最后一個吐出的數(shù)據(jù)是字符串c，這也就是最新的數(shù)據(jù)，然后訂閱下一個對象，下一個對象會依次吐出數(shù)據(jù)，然后跟上一個對象的最新數(shù)據(jù)c結(jié)合，這就得到了上面看到的內(nèi)容
場景三：定制下游數(shù)據(jù) 說明： ?這里就需要啊使用處理函數(shù)了，這個函數(shù)的參數(shù)就是每一個數(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) => {
? ?// 上面只有兩個數(shù)據(jù)源，所以參數(shù)只會前兩個有值
? ?console.log(res1, res2, res3);
?}
);
combinedTimers.subscribe();
=> withLatestFrom操作符 <= 說明： ?這個的作用于combineLatest是類似的，只不過下游推送數(shù)據(jù)只能由一個上游Observable對象驅(qū)動，也就是調(diào)用withLatestFrom的那個Observable對象起到主導(dǎo)數(shù)據(jù)產(chǎn)生節(jié)奏的作用，作為參數(shù)的Observable對象只能貢獻(xiàn)數(shù)據(jù)，不能控制產(chǎn)生數(shù)據(jù)的時機(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)生第一個數(shù)據(jù)的時間為準(zhǔn)，只留下最快 的那一個，當(dāng)然，使用的所有數(shù)據(jù)也是最快的那一個
參數(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操作符 <= 作用： ?在讓?個Observable對象在被訂閱的時候，總是先同步 吐出指定的若?個數(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對象的最后?個 數(shù)據(jù)，將其合并成一個數(shù)組發(fā)射出去
參數(shù)： ?forkJoin(對象 / 數(shù)組)
import { forkJoin, of, timer } from "rxjs";
// 下面會在四秒后返回結(jié)果
forkJoin({
?foo: of(1, 2, 3, 4),
?bar: Promise.resolve(8),
?baz: timer(4000),
}).subscribe((res) => console.log(res));（2）高階Observable 說明： ?簡單理解就是一個Observable中存在Observable，它有一個特點就是高階Observable完結(jié)不代表其里面的Observable完結(jié)
=> concatAll操作符 <= 說明： ?這個操作符針對高階Observable，也是依次訂閱Observable內(nèi)部的Observable取值結(jié)合，訂閱的過程中如果上一個Observable沒有完結(jié)就不會訂閱下一個Observable對象。其他操作可以參照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操作符 <= 說明： ?針對高階Observable，在合并的時候，依次訂閱其內(nèi)部的Observable對象，那個對象有數(shù)據(jù)傳下來，這個數(shù)據(jù)就會傳遞給下游；它可以傳遞一個數(shù)字來限定合并的最大流的個數(shù)。其他操作可以參照merge
參數(shù)： ?mergeAll(數(shù)字)
import { of, mergeAll, repeat } from "rxjs";
// 這里A延遲復(fù)制的時間比B長，所以第二次打印的時候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操作符 <= 說明： ?對高階Observable使用的時候，將數(shù)據(jù)流里面的數(shù)據(jù)一一對應(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(
? ?// 可以接收一個處理函數(shù)，每個參數(shù)對應(yīng)返回值的每一項
? ?zipAll((a, b, c) => {
? ? ?// 這里將參數(shù)打印出來，如果沒有返回值則下游將收不到值
? ? ?console.log(a, b, c);
? ?})
?)
?.subscribe();=> combineLatestAll操作符 <= 說明： ?在處理高階Observable的時候，將其內(nèi)部Observable產(chǎn)生的最新數(shù)據(jù)一一對應(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(
? ?// 可以接收一個處理函數(shù)，每個參數(shù)對應(yīng)返回值的每一項
? ?combineLatestAll((a, b, c) => {
? ? ?// 這里將參數(shù)打印出來，如果沒有返回值則下游將收不到值
? ? ?console.log(a, b, c);
? ?})
?)
?.subscribe();五、輔助類操作符 （1）數(shù)學(xué)類操作符 說明： ?這里介紹的操作符會遍歷上游Observable對象中吐出的所有數(shù)據(jù)才給下游傳遞數(shù)據(jù)， 也就是說，它們只有在上游完結(jié)的時候，才給下游傳遞唯?數(shù)據(jù)。
=> count操作符 <= 作用： ?用于統(tǒng)計上游Observable對象吐出的所有數(shù)據(jù)個數(shù)，所以上游的Observable需要完結(jié)
參數(shù)： ?count(過濾函數(shù))
import { of, interval, count } from "rxjs";
// 可以完結(jié)
of(1000, 1)
?.pipe(
? ? ?// 此時過濾出數(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ù)類型，?如?個對象，那必須指定?個?較這種復(fù)雜類型??的?法，就像上面使用的那樣
=> reduce操作符 <= 說明： ?對上游的每個數(shù)據(jù)進(jìn)行自定義計算，也就是對每一個元素都會調(diào)用一次這個函數(shù)
參數(shù)： ?reduce((累加的值, 當(dāng)前元素的值) => {}, 初始值)
// 計算 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對象的某些條件產(chǎn)生一個新的 Observable對象
=> every操作符 <= 作用： ?它接受一個判定函數(shù)作為參數(shù)，如果上游所有數(shù)據(jù)都能夠通過這個函數(shù)，那么會返回一個包含true值的Observable，有一個通不過就返回一個包含false值的Observable，在吐出結(jié)果后every產(chǎn)生的Observable會立即完結(jié)；不要對不會完結(jié)的對象使用
參數(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操作符 <= 作用： ?通過一個處理函數(shù)來在上游數(shù)據(jù)中查找滿足條件的數(shù)據(jù)，find會吐出找到的上游的數(shù)據(jù)，findIndex會吐出滿足判定條件的索引，如果找不到find會吐出undefined后完結(jié)，findIndex則會吐出-1后完結(jié)；不要對不會完結(jié)的對象使用
參數(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對象是不是空的，如果在完結(jié)之前沒有 吐出數(shù)據(jù)，它就是空的，此時返回一個包含true 值的Observable，否則返回一個包含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操作符 <= 作用： ?接受一個默認(rèn)值，如果 檢測上游的Observable是空的，則把這個默認(rèn)值傳遞給下游，如果 不是空的就把上游的東西傳遞給下游；如果 不傳但是上游檢測還是空的，下游就會收到一個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）過濾類操作符 說明： ?對上游Observable中所有的數(shù)據(jù)使用判定函數(shù)進(jìn)行操作，決定是否某些元素不能通過進(jìn)入下游，如果對某個元素處理結(jié)果為true，表示能通過，否則就不能通過
=> filter操作符 <= 作用： ?跟JavaScript中的filter使用起來是類似的，只不過這里針對的是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ù)據(jù)滿?判定條件，就會立刻被同步傳給 下游。
=> first操作符 <= 作用： ?過濾出Observable中第一個滿足條件的值，在沒有找到的時候會拋出一個錯誤，如果不想這個錯誤傳遞給下游可以使用第二個默認(rèn)值，它的作用是在沒找到滿足條件的值的時候?qū)⑦@個值傳遞出去。如果不傳參數(shù)則將第一個數(shù)據(jù)返回出去，
參數(shù)： ?filter(過濾函數(shù), 默認(rèn)值)
import { first, of } from "rxjs";
// 找不到結(jié)果拋出錯誤，但是給默認(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é)果拋出錯誤
of(1, 3)
?.pipe(first((x) => x % 2 === 0))
?.subscribe((res) => console.log(res));=> last操作符 <= 說明： ?這個作用與first相反，它是找最后一個 滿足條件的值，使用可以參考 first，這里需要注意 ，使用這個操作符的上游必須完結(jié) ，否則操作符不知道哪一個是最后一個數(shù)據(jù)
參數(shù)： ?filter(過濾函數(shù), 默認(rèn)值)
import { last, interval } from "rxjs";
// 這個Observable不會完結(jié)，自然也不會拿到結(jié)果
interval(1000)
?.pipe(last((x) => x % 2 === 0, 2))
?.subscribe((res) => console.log(res));=> take操作符 <= 作用： ?從上游的數(shù)據(jù)中拿指定個數(shù) 的數(shù)據(jù)，拿完之后就會完結(jié)，并將獲取的數(shù)據(jù)返回
參數(shù)： ?take(需要的個數(shù))
import { interval, of, take } from "rxjs";
// 數(shù)據(jù)不夠拿，那就拿完為止
of("a", "b", "c")
?.pipe(take(4))
?.subscribe((res) => console.log(res));
// 獲取指定個數(shù)的數(shù)據(jù)
interval(1000)
?.pipe(take(4))
?.subscribe((res) => console.log(res));注意： ?上游每產(chǎn)生一個數(shù)據(jù)就會立即傳給下游，也就是同步 操作的
=> takeLast操作符 <= 作用： ?從后往前 獲取指定個數(shù) 的數(shù)據(jù)，之后將數(shù)據(jù)一次性 返回出去之后完結(jié)
參數(shù)： ?takeLast(需要的個數(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對象不會完結(jié)的話，那么是拿不到數(shù)據(jù)的，因為不知道誰是最后一個數(shù)據(jù)
=> takeWhile操作符 <= 說明： ?takeWhile接受?個判定函數(shù)作為參數(shù)，這個判定函數(shù)有兩個參數(shù)，分別代表上游的數(shù)據(jù)和對應(yīng)的序號，takeWhile會吐出上游數(shù)據(jù)，直到判定函數(shù)返回false，只要遇到第一個判定函數(shù)返回false的情況， takeWhile產(chǎn)生的Observable就完結(jié)
參數(shù)： ?takeWhile(判定函數(shù), 布爾值)
// 這里關(guān)注第二個參數(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));注意： ?第二個參數(shù)表示是否將第一次導(dǎo)致判定函數(shù)結(jié)果為false的那個值發(fā)射出去，默認(rèn)是false，表示不發(fā)射，true則表示發(fā)射。
=> takeUntil操作符 <= 說明： ?它接受一個Observable對象，在這個對象沒有吐出數(shù)據(jù)之前，上游的數(shù)據(jù)會直接傳遞給下游，在參數(shù)對象吐出第一個數(shù)據(jù)時，上游的數(shù)據(jù)就不能傳遞給下游了。其次參數(shù)對象出現(xiàn)錯誤的時候，這個錯誤會傳遞給下游，此時上游數(shù)據(jù)也不能傳遞給下游了
參數(shù)： ?takeUntil(Observable對象)
// 假如使用interval創(chuàng)建數(shù)據(jù)，在第三秒的時候停止
import { interval, takeUntil, timer } from "rxjs";
interval(1000)
?.pipe(
? ? ?takeUntil(timer(3000))
?)
?.subscribe((res) => console.log(res));=> skip操作符 <= 作用： ?跳過上游的前n個 值，然后從上游的第n+1個 值開始傳遞給下游，這個操作符不關(guān)心最后一個值是什么，所以 這個操作符的上游不管會不會完結(jié)下游都會有值。
參數(shù)： ?skip(跳過的個數(shù))
import { interval, skip } from "rxjs";
// 跳過前兩個值
interval(1000)
?.pipe(skip(2))
?.subscribe((res) => console.log(res));=> skipLast操作符 <= 作用： ?可以理解成去除上游的最后n個 值，然后將剩下的值傳遞給下游；
參數(shù)： ?skipLast(跳過的n個值)
import { interval, skipLast, of } from "rxjs";
// 一個完結(jié)的對象
of("a", "b", "c")
?.pipe(skipLast(2))
?.subscribe((res) => console.log(res));
// 不會完結(jié)的對象
interval(1000)
?.pipe(skipLast(2))
?.subscribe((res) => console.log(res));注意： ?上游沒有完結(jié)下游依然可以收到數(shù)據(jù)
=> skipWhile操作符 <= 說明： ?它接收一個函數(shù)作為參數(shù)，上游的每一個數(shù)據(jù)都會執(zhí)行這個函數(shù)，只要有一個數(shù)據(jù)在函數(shù)中的返回值是false，那么這個數(shù)據(jù)之前的數(shù)據(jù)都會被過濾調(diào)用，剩下的數(shù)據(jù)會傳遞給下游。
參數(shù)： ?skipWhile(處理函數(shù))
import { interval, skipWhile } from "rxjs";
interval(1000)
?.pipe(skipWhile((val) => val % 2 === 0))
?.subscribe((res) => console.log(res));=> skipUntil操作符 <= 作用： ?用于在一個Observable中跳過一些值，直到另一個Observable發(fā)出了特定的信號或者達(dá)到某種狀態(tài)。
參數(shù)： ?skipUntil(Observable對象)
import { interval, timer, skipUntil } from "rxjs";
// 創(chuàng)建一個每秒發(fā)出一個值的Observable
const source$ = interval(1000);
// 創(chuàng)建一個在5秒后發(fā)出第一個值的Observable
const trigger$ = timer(5000);
// 使用skipUntil操作符，跳過source$的值，直到trigger$發(fā)出第一個值
const example$ = source$.pipe(skipUntil(trigger$));
const subscription = example$.subscribe((val) => console.log(val));（2）有損回壓控制 解釋： ?如果數(shù)據(jù)管道中某一個環(huán)節(jié)處理數(shù)據(jù)的速度跟不上數(shù)據(jù)涌現(xiàn)的速度，上游無法把數(shù)據(jù)推送給下游，就會在緩沖區(qū)中積壓數(shù)據(jù)，這就相當(dāng)于對上游施加了壓力，這就是RxJS世界中的回壓。
處理： ?造成這種現(xiàn)象的原因是數(shù)據(jù)管道中某個環(huán)節(jié)數(shù)據(jù)涌?的速度超過了處理速度，那么，既然處理不過來，干脆就舍棄掉某些涌現(xiàn)的數(shù)據(jù)，這種方式稱為有損回壓控制
可選的調(diào)度器： => throttleTime操作符 <= 說明： ?在一個時間范圍內(nèi)，上游傳遞給下游的數(shù)據(jù)只能傳遞一個；這里參數(shù)如果只傳一個，其它值都會使用默認(rèn)值；
參數(shù)： ?throttleTime(時間范圍, 調(diào)度器, 可選參數(shù)對象)
import {
? ? ? ?interval,
? ? ? ?throttleTime,
? ? ? ?asyncScheduler
} from "rxjs";
// 這里每隔1s產(chǎn)生一個數(shù)字
interval(1000)
?.pipe(
? ? ?throttleTime(
? ? ? ? ? ? ? ? ? ? ?2000,
? ? ? ? ? ? ? ? ? ? ?asyncScheduler,
? ? ? ? ? ? ? ? ? ? ?// trailing為true時產(chǎn)生的結(jié)果是：2、4、6...
? ? ? ? ? ? ? ? ? ? ?// leading為true時產(chǎn)生的結(jié)果是：3、6、9...
? ? ? ? ? ? ? ? ? ? ?{ leading: false, trailing: true }
? ? ? ? ? ? ? ? ?)
? ? ?)
?.subscribe((res) => console.log(res));=> debounceTime操作符 <= 說明： ?在一個時間范圍內(nèi)，一直有數(shù)據(jù)產(chǎn)生一直不會將數(shù)據(jù)傳遞給下游，只有在這個時間外產(chǎn)生的第一個數(shù)據(jù)才會傳遞給下游；所以產(chǎn)生數(shù)據(jù)的間隔需要大于這個時間范圍才可以
參數(shù)： ?throttleTime(時間范圍, 調(diào)度器)
import { interval, debounceTime, asyncScheduler } from "rxjs";
// 這里的值如果比2000還小那么就不會有數(shù)據(jù)打印出來
interval(4000)
?.pipe(debounceTime(2000, asyncScheduler))
?.subscribe((res) => console.log(res));=> throttle和debounce操作符 <= 作用： ?這兩個都是使用Observable中的數(shù)據(jù)來控制流量，區(qū)別 在于時機(jī)不同而已
參數(shù)： ?throttle(處理函數(shù), 可選參數(shù)對象)
參數(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)生第一個數(shù)據(jù)0的時候，throttle就和throttleTime一樣，毫不 猶豫地把這個數(shù)據(jù)0傳給了下游，在此之前會將這個數(shù)據(jù)0作為參數(shù)調(diào)用 durationSelector，然后訂閱durationSelector返回的Observable對象，在這個 Observable對象產(chǎn)生第一個對象之前，所有上游傳過來的數(shù)據(jù)都會被丟棄，于是，source$產(chǎn)生的數(shù)據(jù)1就被丟棄了，因為durationSelector返回的 Observable對象被訂閱之后2000毫秒才會產(chǎn)生數(shù)據(jù)。 這個過程，相當(dāng)于throttle每往下游傳遞一個數(shù)據(jù)，都關(guān)上了上下游之間閘門，只有當(dāng)durationSelector產(chǎn)生數(shù)據(jù)的時候才打開這個閘門。到了2000毫秒的時刻，durationSelector第二次被調(diào)用產(chǎn)生的Observable對象終于產(chǎn)生了多個數(shù)據(jù)，閘門被打開，source$產(chǎn)生的第三個數(shù)據(jù)2正好趕上，被 傳遞給了下游，同時關(guān)上閘門，這時候throttle會立刻退訂上一次 durationSelector返回的Observable對象，重新將數(shù)據(jù)2作為參數(shù)調(diào)用 durationSelector來獲得一個新的Observable對象，這個新的Observable對象產(chǎn)生數(shù)據(jù)的時候，閘門才會再次打開?？梢?，durationSelector產(chǎn)生Observable對象只有第一個產(chǎn)生的數(shù)據(jù)會有作用，而且這個數(shù)據(jù)的產(chǎn)生時機(jī)是關(guān)鍵，至于這個數(shù)據(jù)是個什么值不重要。
=> auditTime和audit操作符 <= 說明： ?這兩個都是在一個時間內(nèi)，將最后一個產(chǎn)生的值發(fā)射出去，其余的值會被忽略掉。它們之間的區(qū)別是一個使用時間范圍管理，一個使用函數(shù)管理
參數(shù)： ?auditTime(時間范圍, 可選參數(shù)對象)
參數(shù)： ?audit(處理函數(shù))
import { interval, auditTime } from "rxjs";
interval(1000)
? ?.pipe(auditTime(3000))
? ?.subscribe(
? ? ? ?(val) => console.log("auditTime:", val)
? ?); ? ? ? ? ?
// 第一個3s：0、1、2、3 --> 三秒末也是四秒初發(fā)出值3
// 第二個3s：4、5、6、7 --> 六秒末也是七秒初發(fā)出值7
// ...理解： ?上面的時間寫3s，所以在第一個3s內(nèi)產(chǎn)生了值0、1、2，在第3s結(jié)束的時候，產(chǎn)生了值3，根據(jù)定義，所以第一個3s發(fā)出的值是3，在物理上，第n秒結(jié)束的時候，也就是第n+1秒開始的時候，所以下一個3s是從第四秒開始，然后這個時間內(nèi)產(chǎn)生4、5、6，第7s結(jié)束的時候，產(chǎn)生值7，將其傳遞給下游...后面的值都是這樣產(chǎn)生的，也就是它發(fā)出一個值傳遞到下游之后，它會等待下一個值到達(dá)，才會開始其計時
=> sampleTime和sample操作符 <= 說明： ?sampleTime的作用是搜尋一個時間范圍內(nèi)的最后一個數(shù)據(jù)，將其傳遞給下游，如果這個時間范圍里面沒有值則不會傳值到下游，然后繼續(xù)下一個時間范圍的搜尋； 而sample有點不同，它的參數(shù)接收一個Observable對象來控制Observable，這個參數(shù)被稱為notifier，當(dāng)notifier產(chǎn)生一個數(shù)據(jù)的時候， sample就從上游拿最后一個產(chǎn)生的數(shù)據(jù)傳給下游。
參數(shù)： ?sampleTime(時間范圍, 調(diào)度器)
參數(shù)： ?sample(observable對象)
interval(1000)
?.pipe(sampleTime(2000))
?.subscribe((res) => console.log("sampleTime:", res));理解： ?上面數(shù)據(jù)是每隔1s產(chǎn)生一個，然后我搜尋時間范圍是2s，第一個2s，產(chǎn)生值0、1，將1傳遞出去，繼續(xù)第二個2s的搜尋，產(chǎn)生值2、3，將3傳遞出去...以此類推
（3）去重 => distinct操作符 <= 作用： ?上游同樣的數(shù)據(jù)只有第一次產(chǎn)生時會傳給下游，其余的都被舍棄掉了，判斷是否相等使用的是===
參數(shù)： ?distinct(一個函數(shù)來定制需要對比什么屬性, 一個Observable對象用于清空數(shù)據(jù))
場景一： 基本使用 import { distinct, of } from "rxjs";
of(1, 3, 2, 5, 7, 1, 2)
?.pipe(distinct())
?.subscribe((res) => console.log(res));
場景二： ?對對象使用
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));第二個參數(shù)： ?distinct在運作的時候自己會先創(chuàng)建一個集合，里面存放上游的不同數(shù)據(jù)，每次上游傳遞一個數(shù)據(jù)出來就對比集合中是否有元素跟它相等，相等就舍棄，如果上游數(shù)據(jù)無限多切都是不同的，那么這個集合就會有無限的數(shù)據(jù)在里面，這就存在數(shù)據(jù)壓力，為了解決這個問題，可以使用第二個可選參數(shù)，當(dāng)這個Observable對象產(chǎn)生數(shù)據(jù)的時候，這個集合中的數(shù)據(jù)就會被清空。
=> 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表示上一個值，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)的那?個數(shù)據(jù)，如果找不到，則拋出一個錯誤事件，如果不想出現(xiàn)錯誤，可以使用第二個參數(shù)，在找不到的時候，會將第二個參數(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操作符 <= 作用： ?檢查上游是否只有一個滿足對應(yīng)條件的數(shù)據(jù)，如果答案為是，就向下游傳遞這個數(shù)據(jù)；如果答案為否，就向下游傳遞一個異常
參數(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)化類操作符之后，就會變成f（A）、f（B）、f（C）、f（D）的序列，其中f是一個函數(shù)，作用于上游數(shù)據(jù)之后，產(chǎn)生的就是傳給下游新的數(shù)據(jù)
=> map操作符 <= 說明： ?它接受一個函數(shù)作為參數(shù)，這個函數(shù)通常稱為project，指定了數(shù)據(jù)映射的邏輯 ，每當(dāng)上游推下來一個數(shù)據(jù)，map就把這個數(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）無損回壓控制 說明： ?把上游在一段時間內(nèi)產(chǎn)生的數(shù)據(jù)放到一個數(shù)據(jù)集合中，當(dāng)時機(jī)合適時，把緩存的數(shù)據(jù)匯聚到一個數(shù)組或者Observable對象傳給下游，這就是無損回壓控制
=> windowTime和bufferTime操作符 <= 作用： ?用一個參數(shù)來指定產(chǎn)生緩沖窗口的時間間隔，以此緩存上游的數(shù)據(jù)
參數(shù)： ?windowTime(劃分區(qū)塊間隔, 內(nèi)部區(qū)塊開始間隔, 最多緩存數(shù)據(jù)個數(shù))
參數(shù)： ?bufferTime(劃分區(qū)塊間隔, 內(nèi)部區(qū)塊開始間隔, 最多緩存數(shù)據(jù)個數(shù))
場景一： 基本使用 import { timer, windowTime } from "rxjs";
const source$ = timer(0, 1000);
const result$ = source$.pipe(windowTime(4000));理解： ?windowTime的參數(shù)是4000，也就會把時間劃分為連續(xù)的4000毫秒長度區(qū)塊，在每個時間區(qū)塊中，上游傳下來的數(shù)據(jù)不會直接送給下游，而是在該時間區(qū)塊的開始就新創(chuàng)建一個Observable對象推送給下游，然后在這個時間區(qū)塊內(nèi)上游產(chǎn)生的數(shù)據(jù)放到這個新創(chuàng)建的Observable對象中。在每個4000毫秒的時間區(qū)間內(nèi)，上游的每個數(shù)據(jù)都被傳送給對應(yīng)時間區(qū)間的內(nèi)部Observable對象中，當(dāng)4000毫秒時間一到，這個區(qū)間的內(nèi)部Observable對象就會完結(jié)，將結(jié)果打印出來會發(fā)現(xiàn)控制臺每隔1000毫秒打印一個數(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對象，其中的數(shù)據(jù)是數(shù)組形式， bufferTime會把時間區(qū)塊內(nèi)的數(shù)據(jù)緩存，在時間區(qū)塊結(jié)束的時候把所有緩存的數(shù)據(jù)放在一個數(shù)組再傳給下游，在控制臺你會看見每隔4秒打印一個數(shù)組，因此bufferTime把上游數(shù)據(jù)傳遞出去是需要延遲的
場景二： 第二個參數(shù) 作用： ?指定每個時間區(qū)塊開始的時間間隔。
import { timer, windowTime } from "rxjs";
const source$ = timer(0, 1000);
source$.pipe(windowTime(4000, 2000)).subscribe();理解： ?windowTime使用第二個參數(shù)200之后，產(chǎn)生內(nèi)部Observable的頻率更高了，每200毫秒就會產(chǎn)生一個內(nèi)部Observable對象， 而且各內(nèi)部Observable對象中的數(shù)據(jù)會重復(fù)，例如數(shù)據(jù)2和3就同時出現(xiàn)在第一個和第二個內(nèi)部Observable對象中
import { bufferTime, timer } from "rxjs";
const source$ = timer(0, 1000);
source$
? ?.pipe(bufferTime(4000, 2000, 2))
? ?.subscribe(console.log);理解： ?對于bufferTime，因為需要緩存上游數(shù)據(jù)，不管參數(shù)設(shè)定的數(shù)據(jù)區(qū)間有多短，都無法預(yù)期在這段時間內(nèi)上游會產(chǎn)生多少數(shù)據(jù)，如果上游在短時間內(nèi)爆發(fā)出很多數(shù)據(jù)，那就會給bufferTime很大的內(nèi)存壓力，為了防止出現(xiàn)這種情況可以使用第三個可選參數(shù)來指定每個時間區(qū)間內(nèi)緩存的最多數(shù)據(jù)個數(shù)。
注意： ?如果第一個參數(shù)比第二個參數(shù)大，那么就有可能出現(xiàn)數(shù)據(jù)重復(fù)，如果第二個參數(shù)比第一個參數(shù)大，那么就有可能出現(xiàn)上游數(shù)據(jù)的丟失。之所以說“有可能”，是因為丟失或者重疊的時間區(qū)塊中可能上游沒有產(chǎn)生數(shù)據(jù)，所以也就不會引起上游數(shù)據(jù)的丟失和重復(fù)。從這個意義上說來，windowTime和bufferTime如果用上了第二個參數(shù)，也未必是“止損”的回壓控制。
=> windowCount和bufferCount操作符 <= 作用： ?根據(jù)數(shù)據(jù)個數(shù)來決定內(nèi)部的一個Observabe需要保存多少數(shù)據(jù)。
參數(shù)： ?windowCount(時間區(qū)間長度, 隔幾個數(shù)據(jù)重新開一個區(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還支持可選的第二個參數(shù)，如果不使用第二個參數(shù)，那么所有的時間區(qū)間沒有重疊部分；如果使用了第二個參數(shù)，那么第二個參數(shù)依然是時間區(qū)間的長度，但是每間隔第二個參數(shù)毫秒數(shù)，就會新開一個時間區(qū)間
說明： ?對于bufferCount，和windowCount一樣，區(qū)別只是傳給下游的是緩存數(shù)據(jù)組成的數(shù)組
=> windowWhen和bufferWhen操作符 <= 說明： ?它們接受一個函數(shù)作為參數(shù)，這個函數(shù)返回一個Observable對象，用于控制上游的數(shù)據(jù)分割，每當(dāng)返回的Observable對象產(chǎn)生數(shù)據(jù)或者完結(jié)時，windowWhen就認(rèn)為是一個緩沖區(qū)塊的結(jié)束，重新開啟一個緩沖窗口。bufferWhen跟這個是類似的
參數(shù)： ?windowWhen(處理函數(shù))
import { timer, windowWhen } from "rxjs";
const source$ = timer(0, 100);
const closingSelector = () => {
?return timer(400);
};
// 被訂閱的時候windowWhen就開始?作，?先開啟?個緩沖
// 窗口，然后?刻調(diào)?closingSelector獲得?個Observable對象，
// 在這個Observable對象輸出數(shù)據(jù)的時候，當(dāng)前的緩沖窗?就關(guān)閉，
// 同時開啟?個新的緩沖窗口，然后再次調(diào)?closingSelector
// 獲得?個Observable對象
source$.pipe(windowWhen(closingSelector));=> windowToggle和bufferToggle操作符 <= 說明： ?利?Observable來控制緩沖窗口的開和關(guān)。它需要兩個參數(shù)，第一個參數(shù)是一個Observable對象，當(dāng)產(chǎn)生一個數(shù)據(jù)，代表一個緩沖窗口的開始；同時，第二個參數(shù)是一個函數(shù)，它也會被調(diào)用，用來獲得緩沖窗口結(jié)束的通知；其次函數(shù)的參數(shù)是第一個參數(shù)產(chǎn)生的數(shù)據(jù)，這樣就可以由前一個參數(shù)控制緩沖窗口的開始時機(jī)，函數(shù)控制其關(guān)閉時機(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)??個數(shù)據(jù)，所以每400毫秒就會有?個
// 緩沖區(qū)間開始。每當(dāng)opening$產(chǎn)??個數(shù)據(jù)時，closingSelector
// 就會被調(diào)?返回控制對應(yīng)緩沖區(qū)間結(jié)束的Observable對象，
// 如果參數(shù)為偶數(shù)，就會延時200毫秒產(chǎn)??個數(shù)據(jù)，否則就延時100
// 毫秒產(chǎn)??個數(shù)據(jù)
source$.pipe(windowToggle(openings$, closingSelector));=> window和buffer操作符 <= 說明： ?保持一個Observable類型的參數(shù)，稱為notifier$，每當(dāng)notifer$產(chǎn)生一個數(shù)據(jù)，既是前一個緩存窗口的結(jié)束，也是后一個緩存窗口的開始；如果這個Observable完結(jié)了，那么window產(chǎn)生的一階Observable對象也會完結(jié)，buffer也是類似的
參數(shù)： ?window(一個Observable對象)
import { timer, window } from "rxjs";
const source$ = timer(0, 100);
// 一個不會完結(jié)的Observable
const notifer$ = timer(400, 400);
source$.pipe(window(notifer$));import { timer, window } from "rxjs";
const source$ = timer(0, 100);
// 一個會完結(jié)的Observable
const notifer$ = timer(400);
source$.pipe(window(notifer$));（3）高階map 說明： ?傳統(tǒng)map與高階map的區(qū)別在于其函數(shù)參數(shù)的返回值，前者是將一個數(shù)據(jù)映射成另一個數(shù)據(jù) ，而后者是將一個數(shù)據(jù)轉(zhuǎn)變成一個Observable
import { interval, map } from "rxjs";
const source$ = interval(200);
// 這里每個數(shù)據(jù)都會轉(zhuǎn)換成一個包含數(shù)字0、1、2、3、4的
// Observable對象
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)
? ? ? ? ? ?)
? ? ? ?);理解： ?第一個內(nèi)部Observable對象中的數(shù)據(jù)被完整傳遞給了 concatMap的下游，但是，第一個產(chǎn)生的內(nèi)部Observable對象沒有那么快處理，只有到第一個內(nèi)部Observable對象完結(jié)之后，concatMap才會去訂閱第二個內(nèi)部Observable，這樣就導(dǎo)致第二個內(nèi)部Observable對象中的數(shù)據(jù)排在了后面，絕不會和第一個內(nèi)部Observable對象中的數(shù)據(jù)交叉。
=> mergeMap操作符 <= 說明： ?可以理解成mergeMap = map + mergeAll
注意： ?一旦內(nèi)部Observable發(fā)出一個值，它就會立即將該值傳遞給下游觀察者，而不管其他內(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對象優(yōu)先級總是更高，只要有新的內(nèi)部Observable對象產(chǎn)生，就立刻退訂之前的內(nèi)部 Observable對象，改為從最新的內(nèi)部Observable對象拿數(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(一個處理函數(shù)，用于得到數(shù)據(jù)的key值)
機(jī)制： ?對于上游推送下來的任何數(shù)據(jù)，檢查這個數(shù)據(jù)的key值，如果這個key值是第一次出現(xiàn)，就產(chǎn)生一個新的內(nèi)部Observable對象，同時這個數(shù)據(jù)就是內(nèi)部Observable對象的第一個數(shù)據(jù)；如果key值已經(jīng)出現(xiàn)過，就直接把這個數(shù)據(jù)塞給對應(yīng)的內(nèi)部Observable對象
import { groupBy, interval } from "rxjs";
const source$ = interval(200);
source$.pipe(groupBy((val) => val % 2));理解： ?groupBy的函數(shù)參數(shù)取的是參數(shù)除以2的余數(shù)，所以會產(chǎn)生兩個key值：0和1。從彈珠圖中可以看到，0和2屬于第一個內(nèi)部 Observable對象，第一個內(nèi)部Observable對象收納所有key值為0的數(shù)據(jù)，1 和3屬于第二個內(nèi)部Observable對象，因為它們對應(yīng)的key值為1
=> partition操作符 <= 說明： ?partition接受一個判定函數(shù)作為參數(shù)，對上游的每個數(shù)據(jù)進(jìn)行判定，滿足條件的放一個Observable對象，不滿足條件的放到另一個Observable對象，就這樣來分組，它返回的是一個數(shù)組，包含兩個元素，第一個元素是容納滿組判定條件的Observable對象，第二個元素當(dāng)然是不滿足判定條件的Observable對象。
參數(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一般也不會在后面直接使用鏈?zhǔn)秸{(diào)用，需要把結(jié)果以變量存儲，然后分別處理結(jié)果中的兩個Observable對象
（5）累計數(shù)據(jù) => scan操作符 <= 說明： ?與reduce操作符類似，它也有一個求和函數(shù)參數(shù)和一個可選的seed種子參數(shù)作為求和初始值。scan和reduce的區(qū)別在于scan對上游每一個數(shù)據(jù)都會產(chǎn)生一個求和結(jié)果，reduce是對上游所有數(shù)據(jù)進(jìn)行求和，reduce最多只給下游傳遞一個數(shù)據(jù)，如果上游數(shù)據(jù)永不完結(jié)，那reduce也永遠(yuǎn)不會產(chǎn)生數(shù)據(jù)，scan完全可以處理一個永不完結(jié)的上游Observable對象
參數(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ù)都會被調(diào)用，結(jié)果會傳給下游，同時結(jié)果也會由scan保存，作為下一次調(diào)用規(guī)約函數(shù)時的sum參數(shù)
=> mergeScan操作符 <= 說明： ?它在使用的時候跟scan是類似的，不過它的返回值是一個Observable對象
機(jī)制： ?每當(dāng)上游推送一個數(shù)據(jù)下來，mergeScan就調(diào)用一次求和函數(shù)，并且訂閱返回的Observable對象，之后，這個Observable對象會使用類似merge的方式與下游合并，此時mergeScan會記住傳給下游的最后一個數(shù)據(jù)，當(dāng)上游再次推送數(shù)據(jù)下來的時候，就把最后一次傳遞給下游的數(shù)據(jù)作為求和函數(shù)的sum參數(shù)
注意： ?如果mergeScan返回一個復(fù)雜或者不會完結(jié)的Observable對象，可能會導(dǎo)致上游數(shù)據(jù)和返回的Observable對象會交叉?zhèn)鬟f數(shù)據(jù)給下游，這樣那個值是最后一次傳遞給下游的會很難確定，因此在使用的時候返回的Observable里面包含的值盡量簡單
八、錯誤處理 說明： ?錯誤異常和數(shù)據(jù)一樣，會沿著數(shù)據(jù)流管道從上游向下游流動，流過所有的過濾類或者轉(zhuǎn)化類操作符，最后會觸發(fā)Observer的error方法，不過也不是所有錯誤都交給Observer處理，不然它需要處理的東西就太多了，此時就需要在數(shù)據(jù)管道中處理掉，這里處理異常有兩類方法：恢復(fù)和重試。在實際應(yīng)用中，重試和恢復(fù)往往配合使用，因為重試往往是有次數(shù)限制的，不能無限重試，如果嘗試了次數(shù)上限之后得到的依然是錯誤異常， 還是要用“恢復(fù)”的方法獲得默認(rèn)值繼續(xù)運算。
恢復(fù)：就是本來雖然產(chǎn)生了錯誤異常，但是依然讓運算繼續(xù)下去。最常見的場景就是在獲取某個數(shù)據(jù)的過程中發(fā)生了錯誤，這時候雖然沒有獲得正確數(shù)據(jù)，但是用一個默認(rèn)值當(dāng)做返回的結(jié)果，讓運算繼續(xù)。 重試：就是當(dāng)發(fā)現(xiàn)錯誤異常的時候，認(rèn)為這個錯誤只是臨時的，重新嘗試之前發(fā)生錯誤的操作，寄希望于重試之后能夠獲得正常的結(jié)果，其本質(zhì)是在訂閱上游的同時，退訂上一次訂閱的內(nèi)容 => catchError操作符 <= 作用： ?會在管道中捕獲上游傳遞過來的錯誤
參數(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這個位置會拋出一個錯誤
const error$ = source$.pipe(
?map((value) => {
? ?if (value === 4) {
? ? ?throw new Error("unlucky number 4");
? ?}
? ?return value;
?})
);
// 此時錯誤會被catchError的處理函數(shù)所接收
const catch$ = error$
?.pipe(
? ?// err：被捕獲的錯誤
? ?// caught$：上游緊鄰的那個Observable對象，此處就是指error$了
? ?catchError((err, caught$) => {
? ?
? ? ?// 函數(shù)的返回值是一個Observable對象，用來替代發(fā)生錯誤的那個數(shù)據(jù)，然后傳遞給下游
? ? ?return of(8);
? ?})
?)
?// 錯誤被catchError捕獲處理，所以此處不存在錯誤
?.subscribe(console.log);注意： ?異常函數(shù)的第一個參數(shù)caught$比較有意思，因為它代表的是上游的 Observable對象，如果異常函數(shù)就返回caught$的話，相當(dāng)于讓上游Observable 重新試一遍，所以，catch這個操作符其實不光有恢復(fù)的功能，也有重試的功能
=> retry操作符 <= 第一種參數(shù)： 直接傳一個數(shù)字 說明： ?它可以讓上游的Observable重新試一遍，以達(dá)到重試的目的，它接受一個數(shù)值參數(shù)number，number等于指定重試的次數(shù)， 如果number為負(fù)數(shù)或者沒有number參數(shù)，那么就是無限次retry，直到上游不再拋出錯誤異常為止
參數(shù)： ?retry(重試的次數(shù))
注意： ?retry調(diào)用應(yīng)該有一個正整數(shù)的參數(shù)，也就是要指定有限次數(shù)的重試，否則，很可能陷入無限循環(huán)，畢竟被重試的上游Observable只是有可能重試成功，意思就是也有可能重試不成功，如果真的運氣不好就是重試不成功，也真沒有必要一直重試下去，因為retry通常要限定重試次數(shù)，所以retry通常也要和catch配合使用，重試只是增加獲得成功結(jié)果的概率，當(dāng)重試依然沒有結(jié)果的時候，還是要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ù)： 傳一個配置對象 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);
? ? ? ?// 如果這里沒有返回值，下面只會出現(xiàn)一次重復(fù)
? ? ? ?return of(1000);
? ? ?},
? ?}),
? ?catchError((err, caught$) => {
? ? ?return of(8);
? ?})
?)
?.subscribe();=> finalize操作符 <= 說明： ?它接受一個回調(diào)函數(shù)作為參數(shù)，上游無論是完結(jié)還是出現(xiàn)錯誤這個函數(shù)都會執(zhí)行，只不過在一個數(shù)據(jù)流中只會作用一次，同時這個函數(shù)也無法影響數(shù)據(jù)流。
九、多播 說明： ?多播就是讓一個數(shù)據(jù)流的內(nèi)容被多個Observer訂閱
（1）數(shù)據(jù)流的關(guān)系 說明： ?這里指的是Observable和Observer的關(guān)系，可以理解成前者播放內(nèi)容，后者接受內(nèi)容，播放的形式有單播、廣播和多播
理解概念： （2）Subject 承上啟下： ?根據(jù)第一部分對兩種Observable的理解不難得到Cold Observable實現(xiàn)的是單播，Hot Observable實現(xiàn)的多播
問題： ?如何把Cold Observable變成Hot Observable呢
解決： ?在函數(shù)式編程的世界里，有一個要求是保持不可變性，所以，要把一個Cold Observable對象轉(zhuǎn)換成一個Hot Observable對象，并不是去改變這個Cold Observable對象本身，而是產(chǎn)生一個新的Observable對象，包裝之前Cold Observable對象，這樣在數(shù)據(jù)流管道中，新的Observable對象就成為了下游，想要Hot數(shù)據(jù)源的Observer要訂閱的是這個作為下游的Observable對象，所以此時需要一個中間人來完成轉(zhuǎn)化，這個中間人就是Subject
中間人的職責(zé)： => 雙重身份 <= 說明： ?這里說的是它具有具Observable和Observer的性質(zhì)，雖然?個Subject對象是一個Observable，但是這兩個之間存在區(qū)別，區(qū)別在于Subject是存在記憶的，也就是它能夠記住有哪些Observer訂閱了自己，Subject有狀態(tài)，這個狀態(tài)就是所有Observer的列表，所以，當(dāng)調(diào)用Subject的next函數(shù)時，才可以把消息通知給所有的Observer
import { Subject } from "rxjs";
const subject = new Subject();
// 1號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，這個消息只有1號Observer響應(yīng)，
// 因為當(dāng)前只有?個Observer。同時因為next（1）在2號Observer
// 加?之前執(zhí)?，所以2號Observer沒有接收到1
subject.next(1);
// 2號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")
);
// 這時候調(diào)?subject的next?法推送數(shù)據(jù)2，subject現(xiàn)在知道??
// 有兩個Observer，所以會分別推送消息給1號和2號Observer
subject.next(2);
// subject的1號Observer通過unsubscribe?法退訂
subscription1.unsubscribe();
// 這時候subject知道??只有?個2號Observer，
// 所以，當(dāng)調(diào)?complete?法時，只有2號Observer接到通知
subject.complete();特點： ?后加入的觀察者，并不會獲得加入之前Subject對象上通過next推送的數(shù)據(jù)
實現(xiàn)多播： ?既然Subject既有Observable又有Observer的特性，那么，可以讓一個Subject對象成為一個Cold Observable對象的下游，其他想要Hot數(shù)據(jù)源就可以訂閱這個Subject對象來達(dá)到轉(zhuǎn)換的目的，以此完成多播的操作。
=> 不能重復(fù)使用 <= 說明： ?Subject對象也是一個Observable對象，但是因為它有完結(jié)的狀態(tài)，所以不像Cold Observable對象一樣每次被subscribe都是一個新的開始，正因為如此，Subject對象是不能重復(fù)使用的，所謂不能重復(fù)使用，指的是一個 Subject對象一旦被調(diào)用了complete或者error函數(shù)，那么，它作為Observable 的生命周期也就結(jié)束了，后續(xù)還想調(diào)用這個Subject對象的next函數(shù)傳遞數(shù)據(jù)給下游，會沒有任何反應(yīng)。
import { Subject } from "rxjs";
const subject = new Subject();
// ?先1號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號Observer也成為subject的下游，但是，這時候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)?，不會傳遞給2號Observer，
// 也不會傳遞給1號Observer，但是可以獲取subject的complete通知，
// 可以這樣認(rèn)為，當(dāng)?個Subject對象的complete函數(shù)被調(diào)?之后，
// 它暴露給下游的Observable對象就是?個由empty變量產(chǎn)?的直接
// 完結(jié)的Observable對象
subject.next(3);注意： ?在Subject的生命周期結(jié)束之后，再次調(diào)用next方法沒有任何反應(yīng)，也不會拋出錯誤，這樣可能會認(rèn)為上游所有數(shù)據(jù)都傳遞成功了，這是不合理的，由于Subject是一個Observable，那么它就會存在一個unsubscribe的方法，表示它已經(jīng)不管事了，再次調(diào)用其next方法就會報錯，所以可以像下面這樣達(dá)到警示的目的。
import { Subject, interval, take } from "rxjs";
// tick$會間隔?秒鐘吐出數(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秒的時候subject的unsubscribe函數(shù)被調(diào)?，
// 所以，2秒以后的時間，tick$還要調(diào)?subject的
// next就會拋出?個錯誤異常
setTimeout(() => {
?subject.unsubscribe();
}, 1500);=> 多個上游 <= 說明： ?理論上 可以用一個Subject合并多個Observable的數(shù)據(jù)流，但是這樣做并不合適，原因 在于任何一個上游數(shù)據(jù)流的完結(jié)或者出錯信息都可以終結(jié)Subject對象的生命。
import { Subject, interval, take, map } from "rxjs";
// 這兩個數(shù)據(jù)流都是通過interval產(chǎn)?的Cold Observable對象，
// 每隔?秒鐘吐出?個整數(shù)，然后利?map轉(zhuǎn)化為間隔?秒鐘吐出
// ?個固定的字符串，利?take只從兩個數(shù)據(jù)流中分別拿兩個數(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$每隔?秒鐘吐出?個a字符串，吐出兩個之后完結(jié)，
// tick2$同樣每隔?秒鐘吐出?個字符串，只不過吐出的是b，
// 同樣是吐出兩個之后完結(jié)。因為subject訂閱了tick1$和tick2$，
// 所以理論上結(jié)果應(yīng)該是下面這八個值，但其實并不是
// 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個數(shù)據(jù)之后就會調(diào)用下游的complete函數(shù)，也就是調(diào)用subject的complete函數(shù)，此時它已經(jīng)完結(jié)，后續(xù)的next的方法是沒有效果的，這也是為什么第二個b不會有效果的原因。
=> 錯誤處理 <= 說明： ?如果Subject有多個Observer，并且Subject的某個下游數(shù)據(jù)流產(chǎn)生了一個錯誤異常，而且這個錯誤異常沒有被Observer處理，那這個Subject其他的Observer都會失敗，為了避免這種情況的發(fā)生，每有一個Observer的時候，就需要給它一個處理錯誤的地放就可以解決這個問題了。
十、調(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");解釋： ?因為range是同步輸出數(shù)據(jù)，所有當(dāng)Observer添加之后，會把所有數(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í)行，因為range數(shù)據(jù)的吐出是在after subscribe字符串之后的，那么什么是Scheduler呢？
RxJS中定義Scheduler： （2）內(nèi)置的Scheduler 調(diào)度器 說明 null 默認(rèn)不使用，代表同步執(zhí)行的情況 queueScheduler 利用隊列實現(xiàn)，用于迭代操作 asapScheduler 在當(dāng)前工作之后，下個工作之前執(zhí)行，用于異步轉(zhuǎn)換 asyncScheduler 用于基于時間的操作 animationFrameScheduler 用于創(chuàng)建流暢的瀏覽器動畫
（3）支持Scheduler的操作符 => observeOn操作符 <= 作用： ?根據(jù)上游的Observable對象產(chǎn)生出一個新的Observable對象出來，讓這個新的Observable對象吐出的數(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對象的數(shù)據(jù)推送節(jié)奏，并不能改變上游Observable對象所使用的Scheduler
=> subscribeOn操作符 <= 說明： ?這個跟observeOn的區(qū)別在于前者是控制什么時候訂閱Observable對象，而后者是控制Observable對象何時往下游推送數(shù)據(jù)，使用和參數(shù)是類似的。
400 186 1886
