RxJS教程

時間  2019-11-08
標籤 rxjs 教程 简体版
原文   原文鏈接

入門

基本概念：javascript

  • Observable（可觀察對象）： 表示一個概念，這個概念是一個可調用的將來值或事件的集合。
  • Observer（觀察者）： 一個回調函數的集合，它知道如何去監聽由Observable提供的值。
  • **Subscription（訂閱）：**表示Observable的執行，它主要用於取消Observable的執行。
  • Operator(操做符)： 採用函數式編程風格的純函數，使用像map、filter、concat、flatMap、等這樣的操做符來處理集合。
  • Subject（主體）： 至關於EventEmitter，而且是將值或事件多路推送給多個Observer的惟一方式。
  • Schdulers（調度器）：：用來控制併發而且是中央集權的調度員，容許咱們在發生計算時進行協調，例如setTimeout或requestAnimationFrame 或其餘。

Observable（可觀察對象）

Observables 是多個值的惰性推送集合。它填補了下面表格中的空白：java

單個值 多個值
拉取 Function
推送 Promise

拉取（Pull）vs. 推送（Push)

拉取和推送是兩種不一樣的協議，用來描述數據生產者如何與數據消費者進行通訊的。編程

拉取？ 由消費者來決定什麼時候從生產者那接收數據，生產者自己不知道數據什麼時候交付到消費者手中的。緩存

每一個Javascript函數都是拉取體系。函數式數據的生產者，調用該函數的代碼經過從函數調用中取出一個單個返回值來對該函數進行消費。併發

生產者 消費者
拉取 被動的: 當被請求時產生數據。
推送 主動的: 按本身的節奏產生數據。

推送？ 由生產者來決定什麼時候吧數據發給消費者。消費者自己不知道什麼時候後接受數據異步

Promise是最多見的推送體系類型。Promise(生產者) 將一個解析過的值傳遞給已註冊的回調函數(消費者)，但不一樣於函數的是，由 Promise 來決定什麼時候把值「推送」給回調函數。函數式編程

RxJS引入了Observables，一個新的javascript推送體系。Observable是多個值得生產者，並將值推送給觀察者（消費者）函數

  • Function 是惰性的評估運算，調用時會同步地返回一個單一值
  • Generator 是惰性的評估運算，調用時會同步地返回零到無限多個值
  • Promise 是最終可能返回一個值得運算
  • Observable 是惰性評估運算，它能夠從它被調用的時刻起或異步地返回零到無限多個值。

Observable 剖析

  • 建立Observable
  • 訂閱Observable
  • 執行Observable
  • 清理Observable

建立Observable工具

RX.Observable.create 是Observable構造函數的別名，它接收一個參數subscribe函數。單元測試

// 生產者
var observable = RX.Observable.create(function subscribe(observer){
	var id = setInterval(()=>{
		observer.next('hi')
	},1000);
})
複製代碼

Observable可使用create來建立，但一般咱們使用所謂的建立操做符，像of、from、interval、等等

訂閱Observable

// 觀察者
observable.subscribe(x=>console.log(x))
複製代碼

這代表 subscribe 調用在同一 Observable 的多個觀察者之間是不共享的.對 observable.subscribe 的每次調用都會觸發針對給定觀察者的獨立設置。Observable 甚至不會去維護一個附加的觀察者列表。

執行Observable

Observable.create(function subscribe(observer) {...}) 中...的代碼表示 「Observable 執行」，它是惰性運算，只有在每一個觀察者訂閱後纔會執行。隨着時間的推移，執行會以同步或異步的方式產生多個值。

Observable 執行能夠傳遞三種類型的值：

  • "Next" 通知： 發送一個值，好比數字、字符串、對象，等等。
  • "Error" 通知： 發送一個 JavaScript 錯誤 或 異常。
  • "Complete" 通知： 再也不發送任何值。

下面是 Observable 執行的示例，它發送了三個 "Next" 通知，而後是 "Complete" 通知：

var observable = Rx.Observable.create(function subscribe(observer) {
	observer.next(1);
	observer.next(2);
	observer.next(3);
	observer.complete();
});
複製代碼

Observable 嚴格遵照自身的規約，因此下面的代碼不會發送 "Next" 通知 4：

var observable = Rx.Observable.create(function subscribe(observer) {
	observer.next(1);
	observer.next(2);
	observer.next(3);
	observer.complete();
	observer.next(4); // 由於違反規約，因此不會發送
});
複製代碼

在 subscribe 中用 try/catch 代碼塊來包裹任意代碼是個不錯的主意，若是捕獲到異常的話，會發送 "Error" 通知：

var observable = Rx.Observable.create(function subscribe(observer) {
	try {
		observer.next(1);
		observer.next(2);
		observer.next(3);
		observer.complete();
	} catch (err) {
		observer.error(err); // 若是捕獲到異常會發送一個錯誤
	}
});
複製代碼

清理Observable

由於每一個執行都是其對應觀察者專屬的，一旦觀察者完成接收值，它必需要一種方法來中止執行，以免浪費計算能力或內存資源。

當調用了 observable.subscribe ，觀察者會被附加到新建立的 Observable 執行中。這個調用還返回一個對象，即 Subscription (訂閱)：

var subscription = observable.subscribe(x => console.log(x));
複製代碼

Subscription 表示進行中的執行，它有最小化的 API 以容許你取消執行。想了解更多訂閱相關的內容，請參見 Subscription 類型。使用 subscription.unsubscribe() 你能夠取消進行中的執行：

var observable = Rx.Observable.from([10, 20, 30]);
var subscription = observable.subscribe(x => console.log(x));
// 稍後：
subscription.unsubscribe();
複製代碼

當你訂閱了 Observable，你會獲得一個 Subscription ，它表示進行中的執行。只要調用 unsubscribe() 方法就能夠取消執行。

Observer (觀察者)

什麼是觀察者？ - 觀察者是由 Observable 發送的值的消費者。觀察者只是一組回調函數的集合，每一個回調函數對應一種 Observable 發送的通知類型：next、error 和 complete 。下面的示例是一個典型的觀察者對象：

var 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);
複製代碼

觀察者只是有三個回調函數的對象，每一個回調函數對應一種 Observable 發送的通知類型。

Subscription(訂閱)

什麼是 Subscription ？ - Subscription 是表示可清理資源的對象，一般是 Observable 的執行。Subscription 有一個重要的方法，即 unsubscribe，它不須要任何參數，只是用來清理由 Subscription 佔用的資源。在上一個版本的 RxJS 中，Subscription 叫作 "Disposable" (可清理對象)。

var observable = Rx.Observable.interval(1000);
var subscription = observable.subscribe(x => console.log(x));
// 稍後：
// 這會取消正在進行中的 Observable 執行
// Observable 執行是經過使用觀察者調用 subscribe 方法啓動的
subscription.unsubscribe();
//Subscription 還能夠合在一塊兒，這樣一個 Subscription 調用 unsubscribe() 方法，可能會有多個 Subscription 取消訂閱 。你能夠經過把一個 Subscription 添加到另外一個上面來作這件事：
var observable1 = Rx.Observable.interval(400);
var observable2 = Rx.Observable.interval(300);

var subscription = observable1.subscribe(x => console.log('first: ' + x));
var childSubscription = observable2.subscribe(x => console.log('second: ' + x));

subscription.add(childSubscription);

setTimeout(() => {
// subscription 和 childSubscription 都會取消訂閱
subscription.unsubscribe();
}, 1000);
複製代碼

Subscriptions 還有一個 remove(otherSubscription) 方法，用來撤銷一個已添加的子 Subscription 。

Subject(主體)

什麼是 Subject？ - RxJS Subject 是一種特殊類型的 Observable，它容許將值多播給多個觀察者，因此 Subject 是多播的，而普通的 Observables 是單播的(每一個已訂閱的觀察者都擁有 Observable 的獨立執行)。

Subject 像是 Observable，可是能夠多播給多個觀察者。Subject 還像是 EventEmitters，維護着多個監聽器的註冊表。

每一個Subject都是Observable -對於Subject，你能夠提供一個觀察者並使用subscribe方法，就能夠開始正常接收值。從觀察者角度而言，它沒法判斷Observable執行來自普通的Observable仍是Subject。

在 Subject 的內部，subscribe 不會調用發送值的新執行。它只是將給定的觀察者註冊到觀察者列表中，相似於其餘庫或語言中的 addListener 的工做方式。

每一個 Subject 都是觀察者。 -Subject 是一個有以下方法的對象： next(v)、error(e) 和 complete() 。要給 Subject 提供新值，只要調用 next(theValue)，它會將值多播給已註冊監聽該 Subject 的觀察者們。

var subject = new Rx.Subject();

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

subject.next(1);
subject.next(2);
複製代碼

由於 Subject 是觀察者，這也就在乎味着你能夠把 Subject 做爲參數傳給任何 Observable 的 subscribe 方法，以下面的示例所展現的：

var subject = new Rx.Subject();

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

var observable = Rx.Observable.from([1, 2, 3]);

observable.subscribe(subject); // 你能夠提供一個 Subject 進行訂閱

// 結果
observerA: 1
observerB: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3
複製代碼

使用上面的方法，咱們基本上只是經過 Subject 將單播的 Observable 執行轉換爲多播的。這也說明了 Subjects 是將任意 Observable 執行共享給多個觀察者的惟一方式。

還有一些特殊類型的 Subject：BehaviorSubject、ReplaySubject 和 AsyncSubject。

多播的 Observables

「多播 Observable」 經過 Subject 來發送通知，這個 Subject 可能有多個訂閱者，然而普通的 「單播 Observable」 只發送通知給單個觀察者。

多播 Observable 在底層是經過使用 Subject 使得多個觀察者能夠看見同一個 Observable 執行。

在底層，這就是 multicast 操做符的工做原理：觀察者訂閱一個基礎的 Subject，而後 Subject 訂閱源 Observable 。下面的示例與前面使用 observable.subscribe(subject) 的示例相似：

var source = Rx.Observable.from([1, 2, 3]);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);

// 在底層使用了 `subject.subscribe({...})`:
multicasted.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
multicasted.subscribe({
  next: (v) => console.log('observerB: ' + v)
});

// 在底層使用了 `source.subscribe(subject)`:
multicasted.connect();
複製代碼

multicast 操做符返回一個 Observable，它看起來和普通的 Observable 沒什麼區別，但當訂閱時就像是 Subject 。multicast 返回的是 ConnectableObservable，它只是一個有 connect() 方法的 Observable 。

connect() 方法十分重要，它決定了什麼時候啓動共享的 Observable 執行。由於 connect() 方法在底層執行了 source.subscribe(subject)，因此它返回的是 Subscription，你能夠取消訂閱以取消共享的 Observable 執行。

引用計數

手動調用 connect() 並處理 Subscription 一般太笨重。一般，當第一個觀察者到達時咱們想要自動地鏈接，而當最後一個觀察者取消訂閱時咱們想要自動地取消共享執行。

請考慮如下示例，下面的列表概述了 Subscriptions 發生的通過：

  1. 第一個觀察者訂閱了多播 Observable
  2. 多播 Observable 已鏈接
  3. next 值 0 發送給第一個觀察者
  4. 第二個觀察者訂閱了多播 Observable
  5. next 值 1 發送給第一個觀察者
  6. next 值 1 發送給第二個觀察者
  7. 第一個觀察者取消了多播 Observable 的訂閱
  8. next 值 2 發送給第二個觀察者
  9. 第二個觀察者取消了多播 Observable 的訂閱
  10. 多播 Observable 的鏈接已中斷(底層進行的操做是取消訂閱)

要實現這點，須要顯式地調用 connect()，代碼以下：

var source = Rx.Observable.interval(500);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);
var subscription1, subscription2, subscriptionConnect;

subscription1 = multicasted.subscribe({
  next: (v) => console.log('observerA: ' + v)
});
// 這裏咱們應該調用 `connect()`，由於 `multicasted` 的第一個
// 訂閱者關心消費值
subscriptionConnect = multicasted.connect();

setTimeout(() => {
  subscription2 = multicasted.subscribe({
    next: (v) => console.log('observerB: ' + v)
  });
}, 600);

setTimeout(() => {
  subscription1.unsubscribe();
}, 1200);

// 這裏咱們應該取消共享的 Observable 執行的訂閱，
// 由於此後 `multicasted` 將再也不有訂閱者
setTimeout(() => {
  subscription2.unsubscribe();
  subscriptionConnect.unsubscribe(); // 用於共享的 Observable 執行
}, 2000);
複製代碼

若是不想顯式調用 connect()，咱們可使用 ConnectableObservable 的 refCount() 方法(引用計數)，這個方法返回 Observable，這個 Observable 會追蹤有多少個訂閱者。當訂閱者的數量從0變成1，它會調用 connect() 以開啓共享的執行。當訂閱者數量從1變成0時，它會徹底取消訂閱，中止進一步的執行。

refCount 的做用是，當有第一個訂閱者時，多播 Observable 會自動地啓動執行，而當最後一個訂閱者離開時，多播 Observable 會自動地中止執行。

var source = Rx.Observable.interval(500);
var subject = new Rx.Subject();
var refCounted = source.multicast(subject).refCount();
var subscription1, subscription2, subscriptionConnect;

// 這裏其實調用了 `connect()`，
// 由於 `refCounted` 有了第一個訂閱者
console.log('observerA subscribed');
subscription1 = refCounted.subscribe({
  next: (v) => console.log('observerA: ' + v)
});

setTimeout(() => {
  console.log('observerB subscribed');
  subscription2 = refCounted.subscribe({
    next: (v) => console.log('observerB: ' + v)
  });
}, 600);

setTimeout(() => {
  console.log('observerA unsubscribed');
  subscription1.unsubscribe();
}, 1200);

// 這裏共享的 Observable 執行會中止，
// 由於此後 `refCounted` 將再也不有訂閱者
setTimeout(() => {
  console.log('observerB unsubscribed');
  subscription2.unsubscribe();
}, 2000);

// 執行結果：
observerA subscribed
observerA: 0
observerB subscribed
observerA: 1
observerB: 1
observerA unsubscribed
observerB: 2
observerB unsubscribed
複製代碼

refCount() 只存在於 ConnectableObservable，它返回的是 Observable，而不是另外一個 ConnectableObservable 。

BehaviorSubject

Subject 的其中一個變體就是 BehaviorSubject，它有一個「當前值」的概念。它保存了發送給消費者的最新值。而且當有新的觀察者訂閱時，會當即從 BehaviorSubject 那接收到「當前值」。

BehaviorSubjects 適合用來表示「隨時間推移的值」。舉例來講，生日的流是一個 Subject，但年齡的流應該是一個 BehaviorSubject 。

在下面的示例中，BehaviorSubject 使用值0進行初始化，當第一個觀察者訂閱時會獲得0。第二個觀察者訂閱時會獲得值2，儘管它是在值2發送以後訂閱的。

var subject = new Rx.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
observerA: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3
複製代碼

ReplaySubject

ReplaySubject 相似於 BehaviorSubject，它能夠發送舊值給新的訂閱者，但它還能夠記錄 Observable 執行的一部分。

ReplaySubject 記錄 Observable 執行中的多個值並將其回放給新的訂閱者。

var subject = new Rx.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); 
複製代碼

除了緩衝數量，你還能夠指定 window time (以毫秒爲單位)來肯定多久以前的值能夠記錄。在下面的示例中，咱們使用了較大的緩存數量100，但 window time 參數只設置了500毫秒。

var subject = new Rx.ReplaySubject(100, 500 /* windowTime */);

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

var i = 1;
setInterval(() => subject.next(i++), 200);

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

// 輸出:
observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerA: 5
observerB: 3
observerB: 4
observerB: 5
observerA: 6
observerB: 6
...
複製代碼

AsyncSubject

AsyncSubject 是另外一個 Subject 變體，只有當 Observable 執行完成時(執行 complete())，它纔會將執行的最後一個值發送給觀察者。

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();
複製代碼

輸出：

observerA: 5
observerB: 5
複製代碼

AsyncSubject 和 last() 操做符相似，由於它也是等待 complete 通知，以發送一個單個值。

Operators (操做符)

儘管 RxJS 的根基是 Observable，但最有用的仍是它的操做符。操做符是容許複雜的異步代碼以聲明式的方式進行輕鬆組合的基礎代碼單元。

什麼是操做符?

操做符是 Observable 類型上的方法，好比 .map(...)、.filter(...)、.merge(...)，等等。當操做符被調用時，它們不會改變已經存在的 Observable 實例。相反，它們返回一個新的 Observable ，它的 subscription 邏輯基於第一個 Observable 。

操做符是函數，它基於當前的 Observable 建立一個新的 Observable。這是一個無反作用的操做：前面的 Observable 保持不變。

操做符本質上是一個純函數 (pure function)，它接收一個 Observable 做爲輸入，並生成一個新的 Observable 做爲輸出。訂閱輸出 Observable 一樣會訂閱輸入 Observable 。在下面的示例中，咱們建立一個自定義操做符函數，它將從輸入 Observable 接收的每一個值都乘以10：

function multiplyByTen(input) {
  var output = Rx.Observable.create(function subscribe(observer) {
    input.subscribe({
      next: (v) => observer.next(10 * v),
      error: (err) => observer.error(err),
      complete: () => observer.complete()
    });
  });
  return output;
}

var input = Rx.Observable.from([1, 2, 3, 4]);
var output = multiplyByTen(input);
output.subscribe(x => console.log(x));
複製代碼

輸出：

10
20
30
40
複製代碼

注意，訂閱 output 會致使 input Observable 也被訂閱。咱們稱之爲「操做符訂閱鏈」。

實例操做符 vs. 靜態操做符

什麼是實例操做符？ - 一般提到操做符時，咱們指的是實例操做符，它是 Observable 實例上的方法。舉例來講，若是上面的 multiplyByTen 是官方提供的實例操做符，它看起來大體是這個樣子的：

Rx.Observable.prototype.multiplyByTen = function multiplyByTen() {
  var input = this;
  return Rx.Observable.create(function subscribe(observer) {
    input.subscribe({
      next: (v) => observer.next(10 * v),
      error: (err) => observer.error(err),
      complete: () => observer.complete()
    });
  });
}
複製代碼

實例運算符是使用 this 關鍵字來指代輸入的 Observable 的函數。

注意，這裏的 input Observable 再也不是一個函數參數，它如今是 this 對象。下面是咱們如何使用這樣的實例運算符：

var observable = Rx.Observable.from([1, 2, 3, 4]).multiplyByTen();

observable.subscribe(x => console.log(x));
複製代碼

什麼是靜態操做符？ - 除了實例操做符，還有靜態操做符，它們是直接附加到 Observable 類上的。靜態操做符在內部不使用 this 關鍵字，而是徹底依賴於它的參數。

靜態操做符是附加到 Observalbe 類上的純函數，一般用來從頭開始建立 Observalbe 。

最經常使用的靜態操做符類型是所謂的建立操做符。它們只接收非 Observable 參數，好比數字，而後建立一個新的 Observable ，而不是將一個輸入 Observable 轉換爲輸出 Observable 。

一個典型的靜態操做符例子就是 interval 函數。它接收一個數字(非 Observable)做爲參數，並生產一個 Observable 做爲輸出：

var observable = Rx.Observable.interval(1000 /* 毫秒數 */);
複製代碼

建立操做符的另外一個例子就是 create，已經在前面的示例中普遍使用。點擊這裏查看全部靜態操做符列表。

然而，有些靜態操做符可能不一樣於簡單的建立。一些組合操做符多是靜態的，好比 merge、combineLatest、concat，等等。這些做爲靜態運算符是有道理的，由於它們將多個 Observables 做爲輸入，而不只僅是一個，例如：

var observable1 = Rx.Observable.interval(1000);
var observable2 = Rx.Observable.interval(400);

var merged = Rx.Observable.merge(observable1, observable2);
複製代碼

要解釋操做符是如何工做的，文字描述一般是不足以描述清楚的。許多操做符都是跟時間相關的，它們可能會以不一樣的方式延遲(delay)、取樣(sample)、節流(throttle)或去抖動值(debonce)。圖表一般是更適合的工具。彈珠圖是操做符運行方式的視覺表示，其中包含輸入 Obserable(s) (輸入多是多個 Observable )、操做符及其參數和輸出 Observable 。

在彈珠圖中，時間流向右邊，圖描述了在 Observable 執行中值(「彈珠」)是如何發出的。

在下圖中能夠看到解剖過的彈珠圖。

Alt text

在整個文檔站中，咱們普遍地使用彈珠圖來解釋操做符的工做方式。它們在其餘環境中也可能很是有用，例如在白板上，甚至在咱們的單元測試中(如 ASCII 圖)。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程