引言

最近在項(xiàng)目當(dāng)中別的小伙伴使用到了Rxjs,我一眼看上去有點(diǎn)懵，感覺(jué)挺復(fù)雜，挺繞的。于是抓緊補(bǔ)補(bǔ)課，然后就可以和小伙伴們一起交流怎么能優(yōu)雅的使用Rxjs。由于內(nèi)容比較多，會(huì)分為三篇來(lái)講解說(shuō)明

• 初識(shí) RxJS中的 Observable 和 Observer
• 細(xì)說(shuō) RxJS中的 Operators
• 在談 RxJS中的 Subject和Schedulers

概念

RxJS是一個(gè)庫(kù)，可以使用可觀察隊(duì)列來(lái)編寫異步和基于事件的程序的庫(kù)。

RxJS 中管理和解決異步事件的幾個(gè)關(guān)鍵點(diǎn)：

• Observable： 表示未來(lái)值或事件的可調(diào)用集合的概念。
• Observer： 是一個(gè)回調(diào)集合，它知道如何監(jiān)聽(tīng) Observable 傳遞的值。
• Subscription： 表示一個(gè) Observable 的執(zhí)行，主要用于取消執(zhí)行。
• Operators：** 是純函數(shù)，可以使用函數(shù)式編程風(fēng)格來(lái)處理具有map、filter、concat、reduce等操作的集合。
• Subject: 相當(dāng)于一個(gè)EventEmitter，也是將一個(gè)值或事件多播到多個(gè)Observers的唯一方式。
• Schedulers： 是控制并發(fā)的集中調(diào)度程序，允許我們?cè)谟?jì)算發(fā)生在 eg setTimeoutor requestAnimationFrame或者其它上時(shí)進(jìn)行協(xié)調(diào)。

牛刀小試

我們通過(guò)在dom上綁定事件的小案例，感受一下Rxjs的魅力。

• 在dom綁定事件，我們通常這樣處理

document.addEventListener('click', () => console.log('Clicked!'));

用Rxjs創(chuàng)建一個(gè)observable，內(nèi)容如下

import { fromEvent } from 'rxjs';
fromEvent(document, 'click').subscribe(() => console.log('Clicked!'));

• 這時(shí)候我們簡(jiǎn)單升級(jí)一下，需要記錄一下點(diǎn)擊的數(shù)量

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

用Rxjs可以隔離狀態(tài)，

import { fromEvent, scan } from 'rxjs';
fromEvent(document, 'click')
  .pipe(scan((count) => count + 1, 0))
  .subscribe((count) => console.log(`Clicked ${count} times`));

可以看到，我們用到了scan操作符，該操作符的工作方式和數(shù)組的reduce類似，回調(diào)函數(shù)接收一個(gè)值， 回調(diào)的返回值作為下一次回調(diào)運(yùn)行暴露的一個(gè)值。

通過(guò)上面的案例可以看出，RxJS的強(qiáng)大之處在于它能夠使用純函數(shù)生成值。這意味著您的代碼不太容易出錯(cuò)。 通常你會(huì)創(chuàng)建一個(gè)不純的函數(shù)，你的代碼的其他部分可能會(huì)弄亂你的狀態(tài)。

• 這時(shí)候，需求又有變動(dòng)了，要求我們一秒內(nèi)只能有一次點(diǎn)擊

let count = 0;
let rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', () => {
  if (Date.now() - lastClick >= rate) {
    console.log(`Clicked ${++count} times`);
    lastClick = Date.now();
  }
});

使用Rxjs

fromEvent(document, 'click')
  .pipe(
    throttleTime(1000),
    scan((count) => count + 1, 0)
  )
  .subscribe((count) => console.log(`Clicked ${count} times`));

RxJS 有一系列的操作符，可以幫助你控制事件如何在你的 observables 中流動(dòng)。

• 這時(shí)候，我們要每次累計(jì)鼠標(biāo)x的值

let count = 0;
const rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', (event) => {
  if (Date.now() - lastClick >= rate) {
    count += event.clientX;
    console.log(count);
    lastClick = Date.now();
  }
});

使用Rxjs

import { fromEvent, throttleTime, map, scan } from 'rxjs';
fromEvent(document, 'click')
  .pipe(
    throttleTime(1000),
    map((event) => event.clientX),
    scan((count, clientX) => count + clientX, 0)
  )
  .subscribe((count) => console.log(count));

從上面看可以通過(guò)map去轉(zhuǎn)換observables 的值。

Observable

我們先來(lái)寫一個(gè)案例代碼，大家可以猜下它的執(zhí)行順序

import { Observable } from 'rxjs';
const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => {
    subscriber.next(4);
    subscriber.complete();
  }, 1000);
});
console.log('just before subscribe');
observable.subscribe({
  next(x) { console.log('got value ' + x); },
  error(err) { console.error('something wrong occurred: ' + err); },
  complete() { console.log('done'); }
});
console.log('just after subscribe');

可以稍微想一下，正確的輸出結(jié)果

just before subscribe
got value 1
got value 2
got value 3
just after subscribe
got value 4
done

怎么樣，和大家想的結(jié)果一樣嗎，我們來(lái)一下分析一下。

Observable 剖析

Observable 有兩種方式創(chuàng)建，一種是通過(guò)new Observable()，還有一種是通過(guò)Rx.Observable.create()的方式去創(chuàng)建。

Observable 核心的關(guān)注點(diǎn)：

• 創(chuàng)建Observable
• 訂閱Observable
• 執(zhí)行Observable
• 取消Observable

創(chuàng)建Observable

const observable = new Observable(function subscribe(subscriber) {
  const id = setInterval(() => {
    subscriber.next('hi')
  }, 1000);
});

該代碼是創(chuàng)建一個(gè)Observable,然后每隔1s向訂閱者發(fā)送消息。我們看到上邊的回調(diào)函數(shù)是subscribe, 該函數(shù)是描述Observable最重要的部分。

• 訂閱Observable

observable.subscribe(x => console.log(x));

observable中的subscribe中參數(shù)是一個(gè)回調(diào)x => console.log(x),官方叫它Observer,其實(shí)Observer有多種形式，后邊我們會(huì)說(shuō)到，在這里就簡(jiǎn)單理解，Observer 可以去消費(fèi)數(shù)據(jù)，比如，在react中，我們這可以更新?tīng)顟B(tài)數(shù)據(jù)等。

• 執(zhí)行Observable

 subscriber.next(1);   // Next 通知
 subscriber.complete(); // 完成 通知
 subscriber.error(err);  // Error 通知

其實(shí)就是執(zhí)行一個(gè)惰性計(jì)算，可同步可異步，

Observable Execution 可以傳遞三種類型的值：

• Next：發(fā)送數(shù)值、字符串、對(duì)象等。
• Error：發(fā)送 JavaScript 錯(cuò)誤或異常。
• complete：不發(fā)送值。

Next通知是最重要和最常見(jiàn)的類型：它們代表傳遞給訂閱者的實(shí)際數(shù)據(jù)。在 Observable 執(zhí)行期間，Error和complete通知可能只發(fā)生一次，并且只能有其中之一。

• 取消Observable

function subscribe(subscriber) {
  const intervalId = setInterval(() => {
    subscriber.next('hi');
  }, 1000);
  return function unsubscribe() {
    clearInterval(intervalId);
  };
}
const observable = new Observable(subscribe)
const unsubscribe = observable.subscribe({next: (x) => console.log(x)});
// Later:
unsubscribe(); // 取消執(zhí)行

我們有看代碼，創(chuàng)建了一個(gè)每秒輸出一個(gè)hi內(nèi)容的Observable，但在我們的使用場(chǎng)景中，會(huì)有取消改行為，這時(shí)候就需要返回一個(gè)unsubscribe的方法，用于取消。

Observer

我們?cè)谏线叺膱?chǎng)景中也提到了Observer, 但什么是Observer呢，其實(shí)就是數(shù)據(jù)的消費(fèi)者，先回顧一下上面的代碼

observable.subscribe(x => console.log(x));

其實(shí)可以寫成

const observer = {
  next: x => console.log('Observer got a next value: ' + x),
  error: err => console.error('Observer got an error: ' + err),
  complete: () => console.log('Observer got a complete notification'),
};
observable.subscribe(observer);

這樣應(yīng)就比較清晰了，observer只是具有三個(gè)回調(diào)的對(duì)象，每一個(gè)用于Observable 可能傳遞不同類型的通知。注意，observer 對(duì)象中的類型可以不必要全都寫。

其實(shí)observer有許多變種，我們看下它的TS聲明就比較清楚了。

可以直接傳遞一個(gè)observer對(duì)象，或者只傳遞一個(gè)next回調(diào)函數(shù)，在或者傳多個(gè)可選的回調(diào)函數(shù)類型。

結(jié)束語(yǔ)

RxJS不建議大家盲目的去用，一定要有合適的場(chǎng)景，盲目的去用可能會(huì)造成項(xiàng)目的復(fù)雜度會(huì)大幅度的提升。

以上就是RxJS中的Observable和Observer示例詳解的詳細(xì)內(nèi)容，更多關(guān)于RxJS Observable Observer的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論