RxJS 入門及應(yīng)用

RxJS為什么被稱為是函數(shù)響應(yīng)式編程？

函數(shù)式編程（Functional Programming）

函數(shù)式編程的特點是允許把函數(shù)當(dāng)成一個實參或返回值，主要思想是想將復(fù)雜的運算分解成一系列嵌套的函數(shù)，逐層推導(dǎo)，不斷漸進，直至完成運算。常用的數(shù)組方法（map，filter等）就運用了函數(shù)式編程的思想。

const?arr?=?[4,1,5,2,3];
const?newArr?=?arr
??.sort((a,b)?=>?a-b)
??.filter(value?=>?value>2);
console.log(newArr);???//?[3,4,5]

函數(shù)式編程還有一個最重要的特性，那就是純凈性Purity（純函數(shù)）

純函數(shù)是不會產(chǎn)生副作用的函數(shù)，其中輸出完全由輸入決定，也就是說無論調(diào)用多少次，調(diào)用f(x)都會得到相同的結(jié)果。

函數(shù)副作用：指當(dāng)調(diào)用函數(shù)時，除了返回函數(shù)值之外，還對主調(diào)用函數(shù)產(chǎn)生附加的影響。例如修改全局變量（函數(shù)外的變量），修改參數(shù)或改變外部存儲。

純函數(shù)滿足以下兩個條件

①函數(shù)的執(zhí)行過程完全由輸入?yún)?shù)決定，不會受除參數(shù)之外的任何數(shù)據(jù)的影響。

②函數(shù)不會修改任何外部狀態(tài)，比如修改全局變量或傳入的參數(shù)對象。

通常如果創(chuàng)建一個非純函數(shù)，在這個函數(shù)之外使用了共享變量的代碼，會使應(yīng)用狀態(tài)混亂難以維護。

var?count?=?0;
var?button?=?document.querySelector('button');
button.addEventListener('click',?
??()?=>?console.log(`Clicked?${++count}?times`)
);

而使用RxJS，可以將應(yīng)用狀態(tài)隔離出來，不會被外部環(huán)境影響也不會影響外部環(huán)境。

import?{?fromEvent,?scan?}?from?'rxjs';?//?將事件轉(zhuǎn)換成 observable 序列。
var?button?=?document.querySelector('button');
const?example?=?fromEvent(button,?'click').pipe(
??scan(count?=>?count?+?1,?0);?//工作原理與數(shù)組的?reduce?類似,隨著時間的推移進行歸并
);
example.subscribe(count?=>?console.log(`Clicked?${count}?times`));

響應(yīng)式編程（Reactive Programming）

wiki百科中的解釋：

在計算中，響應(yīng)式編程或反應(yīng)式編程（Reactive programming）是一種面向數(shù)據(jù)流和變化傳播的聲明式編程范式。這意味著可以在編程語言中很方便地表達靜態(tài)或動態(tài)的數(shù)據(jù)流，而相關(guān)的計算模型會自動將變化的值通過數(shù)據(jù)流進行傳播。

什么是數(shù)據(jù)流？

數(shù)據(jù)流（data stream）是數(shù)據(jù)在系統(tǒng)內(nèi)傳播的路徑，表示在一定時間范圍內(nèi)發(fā)生的一系列事件。

任何東西都可以是一個 Stream：變量、用戶輸入、網(wǎng)絡(luò)響應(yīng)、定時器、數(shù)據(jù)結(jié)構(gòu)等等。

什么是變化傳播？

在數(shù)據(jù)流傳播的過程中，可能會有一些事件去組合、創(chuàng)建、過濾這些 Streams，從一個舊的stream映射成一個新的stream。我們不需要去輪詢變化，而是對事件進行監(jiān)聽，在執(zhí)行一個事件后，會自動做出相應(yīng)的響應(yīng)，這就是變化傳播。

前端框架與rxjs的結(jié)合

• 前端框架的職責(zé)（例如react）：數(shù)據(jù)與UI視圖的同步，數(shù)據(jù)發(fā)生更新時，視圖隨之更新；

UI?=?f(data);

• 響應(yīng)式編程的職責(zé)（例如rxjs）：聚焦于數(shù)據(jù)，從數(shù)據(jù)的源頭開始，到數(shù)據(jù)的處理變化，再到數(shù)據(jù)流的訂閱，數(shù)據(jù)的消費；

data?=?g(origin?data)

• 兩者關(guān)系看起并不沖突，并且在某些場景下結(jié)合使用可能會為我們帶來便捷，前端框架可以作為響應(yīng)式編程數(shù)據(jù)的一個消費者；

UI?=?f(g(origin?data))

RxJS是用來干什么的？

RxJS是一個用于處理異步事件流的庫，通過使用 observable 序列來編寫異步和基于事件的程序，實際應(yīng)用場景就是把請求封裝成observerable，通過一些基本操作符（map、filter等等）將返回的數(shù)據(jù)處理并且catch錯誤，將異步事件作為集合來處理。RxJS實際上是將開發(fā)過程中遇到的異步（多為異步，同步也可以）操作看為一個事件流，RxJS內(nèi)部封裝了對一個事件流的操作符（創(chuàng)建、轉(zhuǎn)換、組合、過濾、錯誤異常處理等），組合使用這些操作符來以更便利的方式來管理事件。

為什么用RxJS，摘自知乎回答：

思考一下，異步的本質(zhì)是什么？

異步操作和同步操作最大的區(qū)別就是異步有時序。

我們可以把同步操作理解為：數(shù)據(jù)+函數(shù)

那么異步操作就是：數(shù)據(jù)+函數(shù)+時序

Rx就是把時序抽離成一根時間軸，在這根時間軸上進行同步操作，而異步相關(guān)的時序處理就交給Rx提供的各種operator操作符。

所以問題就很簡單了，如果你的應(yīng)用是一個時序密集的應(yīng)用，那么使用Rx能幫你理清復(fù)雜的異步邏輯。反之，如果異步操作之間沒有太多的聯(lián)系，時序分散， 則不那么需要使用Rx。

RxJS中解決異步事件管理的基本概念

1. Observable（可觀察對象）

將一個數(shù)據(jù)流看作一個可觀察對象，表示這個數(shù)據(jù)流變化傳播過程中發(fā)生的一些事件的集合。

拉取和推送是兩種不同的協(xié)議，用來描述數(shù)據(jù)生產(chǎn)者 (Producer) 與數(shù)據(jù)消費者 (Consumer) 如何通信。

1. 拉取體系

js中每個函數(shù)function都屬于拉取體系，函數(shù)來生產(chǎn)數(shù)據(jù)，消費者通過調(diào)用該函數(shù)的代碼來從函數(shù)中獲取單個返回值來對該函數(shù)進行消費，而迭代器Iterator則是消費者調(diào)用iterator.next()來獲取多個返回值進行消費。

拉取的過程中，生產(chǎn)者是一個被動的過程，在消費者請求調(diào)用自己時才產(chǎn)生數(shù)據(jù)，消費者是一個主動的過程，消費者自己來決定何時調(diào)用生產(chǎn)者來獲取收據(jù)。

2. 推送體系

在如今的js中，Promise是最常見的推送體系，Promise作為生產(chǎn)者，將解析過的resolved值傳給消費者注冊過的一個回調(diào)函數(shù)。

推送的過程中，生產(chǎn)者是一個主動的過程，在生產(chǎn)者獲取resolved值的時候，生產(chǎn)者可以決定何時把值推送給消費者，而消費者并不知道什么時候可以從生產(chǎn)者這里獲取到值。在RxJS中，observable也屬于推送體系，并且可以推送一個或多個值。

上面這些術(shù)語有些抽象，舉個??更容易理解什么是Observable

• Observable就像是一個function函數(shù)

Function
function?foo()?{
????console.log('Hello')
????return?'world';
}

const?x?=?foo();
console.log(x);

const?y?=?foo();
console.log(y);

Observable
const?foo?=?Observable.create(function?(observer)?{
????console.log('Hello');
????observer.next('world');
});
//?.subscribe()類似于調(diào)用函數(shù)
foo.subscribe(function?(x)?{
????console.log(x);
});
foo.subscribe(function?(y)?{
????console.log(y);
});

//?控制臺輸出是相同的：?
'Hello'
'world'
'Hello'
'world'

• Observable和function的區(qū)別是什么?

Observable可以隨著時間推移返回（推送）多個值 ，這一點是函數(shù)做不到的。

Function
function?foo()?{
????return?'Hello';
????return?'world';?//?永遠不會執(zhí)行
}

const?a?=?foo();
console.log(a);

//控制臺輸出
'Hello'

Observable
const?foo?=?Observable.create(function?(observer)?{
????observer.next('Hello');
????observer.next('world');
});

foo.subscribe(function?(x)?{
????console.log(x);
});

//?控制臺輸出
'Hello'
'world'

//?也可以異步推送一些值
const?foo?=?Observable.create(function?(observer)?{
????observer.next('Hello')；
????setTimeout(()?=>?{
????????observer.next('rxjs');
????},0)
????observer.next('world');
});

//?控制臺輸出
'Hello'
'world'
'rxjs'

冰墩墩占位

1.1 創(chuàng)建Observable

Observable可以使用Observable.create來創(chuàng)建，但通常我們使用創(chuàng)建操作符^[1]來創(chuàng)建Observable。

1.2 ? 訂閱Observable

訂閱Observable像是調(diào)用函數(shù)，并提供接收數(shù)據(jù)的回調(diào)函數(shù)。

observable.subscribe(value?=>?{
?//?do?something
})

不同觀察者通過subscribe調(diào)用同一observable數(shù)據(jù)不共享。

每一次調(diào)用，等于重新執(zhí)行一遍函數(shù)。

1.3 執(zhí)行Observable

Observable執(zhí)行可以傳遞三種類型的值：

1. Next：推送一個值，可以是任意類型；

2. Error：推送一個錯誤或者異常；

3. Complete：推送一個「已完成」的消息，表明不會再發(fā)送任何值；

next()方法中的值代表要推送給觀察者的實際數(shù)據(jù)，可以執(zhí)行多次；

error()和complete()會在Observable執(zhí)行期間至多執(zhí)行一次，并且只會執(zhí)行其中一個；

Observable.create(observer?=>?{
????try?{
????????observer.next(1);
????????observer.next(2);
????????observer.complete();
????????observer.next(3);?//?前面已經(jīng)通知觀察者已經(jīng)完成，所以這個值不會發(fā)送
????}?catch?(e)?{
????????observer.error(e);?//?捕獲到異常發(fā)送一個錯誤
????}
})

1.4 ? 銷毀Observable執(zhí)行

Observable的執(zhí)行可能會是無限的，通常觀察者希望在一個有限的時間里終止Observable執(zhí)行，以避免浪費計算資源和內(nèi)存消耗。

類似于清除定時器，var timer = setInterval(() => {},1000); clearInterval(timer);

//?調(diào)用subscribe時，觀察者會被附加到新創(chuàng)建的Observable執(zhí)行中，
//?會返回一個對象，即Subscription（訂閱）
var?subscription?=?observable.subscribe();
// Subscription表示正在進行中的執(zhí)行，調(diào)用unsubscribe()來取消observable執(zhí)行；
subscription.unsubscribe();

2. Observer (觀察者)

Observer（觀察者）是一組回調(diào)函數(shù)的集合，每一個回調(diào)函數(shù)對應(yīng)Observable發(fā)送通知的類型：next、error、complete。

const?observer?=?{
????next:?()?=>?{},?//?觀察者接收到next()消息執(zhí)行的回調(diào)函數(shù)
????error:?()?=>?{},?//?觀察者接收到error()消息執(zhí)行的回調(diào)函數(shù)
????complete:?()?=>?{},?//?接收到complete()消息執(zhí)行的回調(diào)函數(shù)
}
// observer中的觀察者可能是部分的，沒有提供某個回調(diào)，observable還是可以執(zhí)行的。
//?方法1：將observer觀察者傳入subscribe
observable.subscribe(observer)
//?方法2：subscribe按順序（next，error，complete)傳入三個回調(diào)函數(shù)
observable.subscribe((value)?=>?{},(error)?=>?{},?()?=>?{})

3. ?Subscription (訂閱)

Subscription是一個可清理資源的對象，代表Observable的執(zhí)行。

基本用處就是使用unsubscribe來釋放資源或取消Observable的執(zhí)行。

4. Subject (主體)

引入一個新的概念，Cold Observable / Hot Observable。

Observable對象就是一個數(shù)據(jù)流，在一個時間范圍內(nèi)推送一系列數(shù)據(jù)。

在只存在一個observer的情況下很簡單，但是對于存在多個observer的場景，會變得復(fù)雜。

假設(shè)一個場景：

兩個observer觀察者A和B訂閱同一個Observable對象，但他們不是同時訂閱，第一個觀察者A訂閱N秒后，第二個觀察者B才訂閱這個Observable對象。并且在這N秒期間,Observable已經(jīng)推送了一些數(shù)據(jù)，那么第二個觀察者B應(yīng)不應(yīng)該收到已經(jīng)被推送給第一個觀察者A的那些數(shù)據(jù)呢？

Selection 1 ：已經(jīng)推送給觀察者A的值就不給B了，B只從訂閱那一時間點接收Observable推送的數(shù)據(jù)就行了。

Selection 2：已經(jīng)推送給觀察者A的值還是要給B，B訂閱時從頭開始獲取Observable推送的數(shù)據(jù)。

RxJS考慮到這兩種不同的場景，讓Observable支持這兩種不同的需求，Selection 1這樣的Observable就是Hot Observable，而Selection 2這樣的Observable就是Cold Observable。

RxJS Subject是一種特殊類型的Observable，允許將值多播給多個觀察者（每個已訂閱的觀察者從訂閱時間點開始接收當(dāng)前Observable推送的值，非獨立），而普通的Observable是單播的（每個已訂閱的觀察者是獨立執(zhí)行Observable的）。

對于多個訂閱Subject的觀察者，subscribe不會重新從頭發(fā)送值，他只是將觀察者注冊到觀察者列表中，后續(xù)有新值發(fā)送的時候，將值多播給觀察者列表中的所有觀察者。

RxJS的四種不同類型Subject

4.1.1 ? BehaviorSubject

BS有一個“當(dāng)前值”的概念，它保存了發(fā)送給觀察者的最后一個值（當(dāng)前值），當(dāng)有新的觀察者訂閱時，會立即接收到“當(dāng)前值”；

而如果用Subject，在觀察者訂閱時，之前已發(fā)送的值不會再發(fā)給觀察者包括最近的一個值，后續(xù)再有值發(fā)送的時候，新注冊的觀察者才會接收到新的值。

var?subject?=?new?BehaviorSubject(0);?//?0是初始值

subject.subscribe({
??next:?(v)?=>?console.log('observerA:?'?+?v)
});

subject.next(1);
subject.next(2);

subject.subscribe({
??next:?(v)?=>?console.log('observerB:?'?+?v)
});

subject.next(3);
//?輸出：
observerA:?0?//line3 ：A訂閱時立即收到當(dāng)前值（初始值）0
observerA:?1?//line7?:?BS推送新的值1，訂閱者A接收到值1
observerA:?2?//line8?:?BS推送新的值2，訂閱者A接收到值2
observerB:?2?//line?10?:?B訂閱時立即收到變化后的當(dāng)前值2
observerA: 3 //line 14：BS推送新的值3，訂閱者A和B一起收到值3
observerB:?3

4.1.2 AsyncSubject

AS只有當(dāng)Observable執(zhí)行完成時【執(zhí)行complete()】，才會將執(zhí)行的最后一個值發(fā)送給觀察者

var?subject?=?new?Rx.AsyncSubject();

subject.subscribe({
??next:?(v)?=>?console.log('observerA:?'?+?v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
??next:?(v)?=>?console.log('observerB:?'?+?v)
});

subject.next(5);
subject.complete();
//?輸出：
//?line?17?執(zhí)行complete()后兩個訂閱者A和B才會收到最后的一個值(5)
observerA:?5
observerB:?5

4.1.3 ReplaySubject

RS類似BS，它可以發(fā)送舊值給新的觀察者，還可以記錄Observable的執(zhí)行的一部分，將Observable執(zhí)行過程中的多個值回放給新的觀察者。

var?subject?=?new?ReplaySubject(3);?//?為新的訂閱者緩沖3個值

subject.subscribe({
??next:?(v)?=>?console.log('observerA:?'?+?v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
??next:?(v)?=>?console.log('observerB:?'?+?v)
});

subject.next(5);
//?輸出
observerA:?1?//?line?7:?RS推送值1，訂閱者A收到值1
observerA:?2?//?line?8:?RS推送值2，訂閱者A收到值2
observerA:?3?//?Line?9:?RS推送值3，訂閱者A收到值3
observerA:?4?//?line?10:?RS推送值4，訂閱者A收到值4
observerB:?2?//?line?12:?新的訂閱者訂閱RS
observerB:?3?//?訂閱時按順序收到了RS緩沖的三個值
observerB:?4
observerA: 5 // line 16：RS推送值5，觀察者A和B收到值5
observerB:?5

Tips
//RS除了可以指定緩沖數(shù)量，還可以指定時間（單位毫秒）來確定多久之前的值要記錄
var subject = new ReplaySubject(3,500)?記錄3個值，500ms前。

5. Operators (操作符)

操作符是允許復(fù)雜的異步代碼以聲明式的方式進行輕松組合的基礎(chǔ)代碼單元。

操作符本質(zhì)就是一個純函數(shù)，當(dāng)操作符被調(diào)用時，不會改變已經(jīng)存在的Observable實例，會基于當(dāng)前Observable創(chuàng)建一個新的Observable。

一個Observable對象代表的是一個數(shù)據(jù)流，實際場景中，產(chǎn)生Observable對象并不是每次都通過直接調(diào)用Observable構(gòu)造函數(shù)來創(chuàng)造數(shù)據(jù)流對象。于現(xiàn)實中復(fù)雜的問題，并不會創(chuàng)造一個數(shù)據(jù)流之后就直接通過 subscribe接上一個Observer，往往需要對這個數(shù)據(jù)流做一系列處理，然后才交給Observer。就像一個管道，數(shù)據(jù)從管道的一段流入，途徑管道各個環(huán)節(jié)，當(dāng)數(shù)據(jù)到達Observer的時候，已經(jīng)被管道操作過，有的數(shù)據(jù)已經(jīng)被中途過濾拋棄掉了，有的數(shù)據(jù)已經(jīng)被改變了原來的形態(tài)，而且最后的數(shù)據(jù)可能來自多個數(shù)據(jù)源，最后Observer只需要處理能夠走到終點的數(shù)據(jù)，而這個數(shù)據(jù)管道就是pipe。

而對于每一個操作符，鏈接的就是上游(upstream)和下游 (downstream)?

Marble diagrams （彈珠圖）

為了能夠解釋流是如何變化的，文字通常不足以能夠描述清楚，我們常常使用彈珠圖來對流的時序變化（操作符的運行方式）進行描述。

一個彈珠圖??：

參考資料

https://cn.rx.js.org/manual/overview.html

https://rxjs-cn.github.io/learn-rxjs-operators/

參考資料

?? 謝謝支持

以上便是本次分享的全部內(nèi)容，希望對你有所幫助^_^

喜歡的話別忘了?分享、點贊、收藏?三連哦~。

歡迎關(guān)注公眾號?趣談前端?收獲大廠一手好文章~