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

牛刀小試

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

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

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

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

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

這時候我們簡單升級一下，需要記錄一下點擊的數量

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操作符，該操作符的工作方式和數組的reduce類似，回調函數接收一個值，回調的返回值作為下一次回調運行暴露的一個值。

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

這時候，需求又有變動了，要求我們一秒內只能有一次點擊

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 中流動。

這時候，我們要每次累計鼠標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));

從上面看可以通過map去轉換observables 的值。

Observable

我們先來寫一個案例代碼，大家可以猜下它的執(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');

可以稍微想一下，正確的輸出結果

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

怎么樣，和大家想的結果一樣嗎，我們來一下分析一下。

Observable 剖析

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

Observable 核心的關注點：

創(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)建一個Observable,然后每隔1s向訂閱者發(fā)送消息。我們看到上邊的回調函數是subscribe, 該函數是描述Observable最重要的部分。

訂閱Observable

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

observable中的subscribe中參數是一個回調x => console.log(x),官方叫它Observer,其實Observer有多種形式，后邊我們會說到，在這里就簡單理解，Observer 可以去消費數據，比如，在react中，我們這可以更新狀態(tài)數據等。

執(zhí)行Observable

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

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

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

Next：發(fā)送數值、字符串、對象等。
Error：發(fā)送 JavaScript 錯誤或異常。
complete：不發(fā)送值。

Next通知是最重要和最常見的類型：它們代表傳遞給訂閱者的實際數據。在 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)建了一個每秒輸出一個hi內容的Observable，但在我們的使用場景中，會有取消改行為，這時候就需要返回一個unsubscribe的方法，用于取消。

Observer

我們在上邊的場景中也提到了Observer, 但什么是Observer呢，其實就是數據的消費者，先回顧一下上面的代碼

observable.subscribe(x =&gt; console.log(x));

其實可以寫成

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);

這樣應就比較清晰了，observer只是具有三個回調的對象，每一個用于Observable 可能傳遞不同類型的通知。注意，observer 對象中的類型可以不必要全都寫。

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