摘要：是的縮寫，起源于，是一個基于可觀測數據流結合觀察者模式和迭代器模式的一種異步編程的應用庫。是基于觀察者模式和迭代器模式以函數式編程思維來實現的。學習之前我們需要先了解觀察者模式和迭代器模式，還要對流的概念有所認識。

RxJS 是 Reactive Extensions for JavaScript 的縮寫，起源于 Reactive Extensions，是一個基于可觀測數據流 Stream 結合觀察者模式和迭代器模式的一種異步編程的應用庫。RxJS 是 Reactive Extensions 在 JavaScript 上的實現。

Reactive Extensions（Rx）是對 LINQ 的一種擴展，他的目標是對異步的集合進行操作，也就是說，集合中的元素是異步填充的，比如說從 Web
或者云端獲取數據然后對集合進行填充。LINQ（Language Integrated Query）語言集成查詢是一組用于 C# 和
Visual Basic 語言的擴展。它允許編寫 C# 或者 Visual Basic 代碼以操作內存數據的方式，查詢數據庫。

RxJS 的主要功能是利用響應式編程的模式來實現 JavaScript 的異步式編程（現前端主流框架 Vue React Angular 都是響應式的開發框架）。

RxJS 是基于觀察者模式和迭代器模式以函數式編程思維來實現的。學習 RxJS 之前我們需要先了解觀察者模式和迭代器模式，還要對 Stream 流的概念有所認識。下面我們將對其逐一進行介紹，準備好了嗎？讓我們現在就開始吧。

觀察者模式又叫發布訂閱模式（Publish/Subscribe），它是一種一對多的關系，讓多個觀察者（Obesver）同時監聽一個主題（Subject），這個主題也就是被觀察者（Observable），被觀察者的狀態發生變化時就會通知所有的觀察者，使得它們能夠接收到更新的內容。

觀察者模式主題和觀察者是分離的，不是主動觸發而是被動監聽。

舉個常見的例子，例如微信公眾號關注者和微信公眾號之間的信息訂閱。當微信用戶關注微信公眾號 webinfoq就是一個訂閱過程，webinfoq負責發布內容和信息，webinfoq有內容推送時，webinfoq的關注者就能收到最新發布的內容。這里，關注公眾號的朋友就是觀察者的角色，公眾號webinfoq就是被觀察者的角色。

示例代碼：

// 定義一個主題類（被觀察者/發布者）
class Subject {
  constructor() {
    this.observers = [];   // 記錄訂閱者（觀察者）的集合
    this.state = 0;        // 發布的初始狀態
  }
  getState() {
    return this.state;
  }
  setState(state) {        
    this.state = state;    // 推送新信息
    this.notify();         // 通知訂閱者有更新了
  }
  attach(observer) {
    this.observers.push(observer);   // 對觀察者進行登記
  }
  notify() {
    // 遍歷觀察者集合，一一進行通知
    this.observers.forEach(observer = {
      observer.update();   
    })
  }
}

// 定義一個觀察者（訂閱）類
class Observer {
  constructor(name, subject) {  
    this.name = name;   // name 表示觀察者的標識
    this.subject = subject;  // 觀察者訂閱主題
    this.subject.attach(this);  // 向登記處傳入觀察者實體
  }
  update() {
    console.log(`${this.name} update, state: ${this.subject.getState()}`);
  }
}

// 創建一個主題
let subject = new Subject();

// 創建三個觀察者： observer$1 observer$2 observer$3
let observer$1 = new Observer("observer$1", subject);
let observer$2 = new Observer("observer$2", subject);
let observer$3 = new Observer("observer$3", subject);

// 主題有更新
subject.setState(1);
subject.setState(2);
subject.setState(3);

// 輸出結果
// observer$1 update, state: 1
// observer$1 update, state: 1
// observer$1 update, state: 1
// observer$2 update, state: 2
// observer$2 update, state: 2
// observer$2 update, state: 2
// observer$3 update, state: 3
// observer$3 update, state: 3
// observer$3 update, state: 3

迭代器（Iterator）模式又叫游標（Sursor）模式，迭代器具有 next 方法，可以順序訪問一個聚合對象中的各個元素，而不需要暴露該對象的內部表現。

迭代器模式可以把迭代的過程從從業務邏輯中分離出來，迭代器將使用者和目標對象隔離開來，即使不了解對象的內部構造，也可以通過迭代器提供的方法順序訪問其每個元素。

了解更多可迭代對象：「JS篇」你不知道的 JS 知識點總結（一）

使用 ES5 創建一個迭代器

//創建一個迭代類，傳入目標對象
function Iterator(container) {
  this.list = container.list;
  this.index = 0;

  //定義私有的next方法，執行迭代
  this.next = function() {
    if(this.hasNext()) { //判斷是否迭代完畢
      return {
        value: this.list[this.index++],
        done: false
      }
    }
    return {value: null, done: true}
  }
  this.hasNext = function() {
    if(this.index >= this.list.length) {
      return false;
    }
    return true;
  }
}

//定義目標對象
function Container(list) {
  this.list = list;
  this.getIterator = function() {
    return new Iterator(this); //用戶返回一個迭代器
  }
}

//調用
var container = new Container([1, 2, 3, 4, 5]);
var iterator = container.getIterator();
iterator.next();  // {value: 1, done: false}
iterator.next();  // {value: 2, done: false}
iterator.next();  // {value: 3, done: false}
iterator.next();  // {value: 4, done: false}
iterator.next();  // {value: 5, done: false}
iterator.next();  // {value: null, done: true}

使用 ES6 構造一個迭代器

class Iterator {
  constructor(container) {
    this.list = container.list;
    this.index = 0;
  }
  next() {
    if(this.hasNext()) {
      return {
        value: this.list[this.index++],
        done: false
      }
    }
    return {value: null, done: true}
  }
  hasNext() {
    if(this.index >= this.list.length) {
      return false;
    }
    return true;
  }
}

class Container {
  constructor(list) {
    this.list = list;
  }
  getIterator() {
    return new Iterator(this);
  }
}

let container = new Container([1, 2, 3, 4, 5]);
let iterator = container.getIterator();

iterator.next();  // {value: 1, done: false}
iterator.next();  // {value: 2, done: false}
iterator.next();  // {value: 3, done: false}
iterator.next();  // {value: 4, done: false}
iterator.next();  // {value: 5, done: false}
iterator.next();  // {value: null, done: true}

使用 ES6 的 Symbol.iterator 創建一個迭代器

var list = [1, 2, 3, 4, 5];
var iterator = list[Symbol.iterator]();

iterator.next();  // {value: 1, done: false}
iterator.next();  // {value: 2, done: false}
iterator.next();  // {value: 3, done: false}
iterator.next();  // {value: 4, done: false}
iterator.next();  // {value: 5, done: false}
iterator.next();  // {value: null, done: true}

通過上邊的示例代碼我們可以得知，我們不了解對象的內部構造，但是可以通過調用迭代器提供的 next() 方法就能順序訪問其每個元素。

在這里可以將一系列的鼠標點擊、鍵盤點擊產生的事件和將要處理的元素集合看作一種流， 流的特點是數據源的本身是無限的，流在管道中傳輸， 并且可以在管道的節點上進行處理， 比如篩選， 排序，聚合等。

流數據源（source）經過數據轉換等中間操作的處理，最后由最終操作得到前面處理的結果，每次轉換原有 Stream 對象不改變，返回一個新的 Stream 對象（可以有多次轉換），這就允許對其操作可以像鏈條一樣排列，變成一個管道。

為了對 stream 有一個更感性的認識，我們說點擊事件可以看作一種 stream，在介紹 RxJS 之前，我們不妨先看一個 RxJS 官網上的例子（列舉官方的例子更能充分體現 RxJS 是基于可觀測數據流 Stream 的）。

通常，注冊一個事件偵聽器是這樣的。

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

使用 RxJS 可以創建一個 observable。

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

結合上邊講到流和設計模式，為了方便我們對 RxJS 有進一步的認識，我們就用代碼自己來實現一個 Obserable。

其實 Observable 實際上就是一個函數，它接收一個Observer 對象作為參數，返回一個函數用來取消訂閱。Observer 對象可以聲明 next、err、complete 方法來處理流的不同狀態。

首先我們定義數據源 Source

// 創建數據源
class Source {
  constructor() {
    this.state = 0;
    this.data = setInterval(() => this.emit(this.state++), 200);
  }
  
  emit(state) {
    const limit = 10;  // 定義數據上限
    if (this.onData) {
      this.onData(state);  // 產生數據
    }
    if (state === limit) {
      if (this.onComplete) {
        this.onComplete();  // 數據終止
      }
      this.destroy();
    }
  }
  
  destroy() {  //停止定時器，清除數據
    clearInterval(this.data);
  }
}

創建一個 Observable

// 創建 Observable
class Observable {
  constructor() {}
  getStream() {  
    return new Source();
  }
  subscribe(observer) {
    // 獲取流數據源
    this.stream = this.getStream();  
    
    // 轉換
    this.stream.onData = (e) => observer.next(e); // 處理流數據
    this.stream.onError = (err) => observer.error(err);  //處理異常
    this.stream.onComplete = () => observer.complete();  //處理流數據終止
    
    // 返回一個函數
    return () => {
      this.stream.destroy();        
    }
  }
}

調用 subscribe 進行訂閱

const observable = new Observable();
//訂閱
let observer = {
  next(data) { console.log(data); },
  error(err) { console.error(err); },
  complete() { console.log("done")}
}
const unsubscribe = observable.subscribe(observer);

輸出結果

我們可以調用 unsubscribe 取消訂閱

//0.5后取消訂閱
setTimeout(unsubscribe, 500);

我們可以看到 Observable 作為生產者與觀察者之間的橋梁，并返回一種方法來解除生產者與觀察者之間的聯系，其中觀察者用于處理時間序列上數據流。

介紹完 RxJS 的一些前置知識點，下面就讓我們一起來認識下什么是 RxJS 吧。

RxJS 中含有兩個基本概念：Observables 與 Observer。Observables 作為被觀察者，是一個值或事件的流集合；而 Observer 則作為觀察者，根據 Observables 進行處理。

Observables 與 Observer 之間的訂閱發布關系(觀察者模式) 如下：

訂閱：Observer 通過 Observable 提供的 subscribe() 方法訂閱 Observable。

發布：Observable 通過回調 next 方法向 Observer 發布事件。

Observable 屬于全新的 push 體系，讓我們先了解下什么是 pull 體系和 push 體系吧。

Pull 和 Push 是兩種不同的協議，描述了數據生產者如何與數據消費者進行通信。

在 Pull 體系中，數據的消費者決定何時從數據生產者那里獲取數據，而生產者自身并不會意識到什么時候數據將會被發送給消費者。

每一個 JavaScript函數都是一個 Pull 體系，函數是數據的生產者，調用函數的代碼通過 "拉出" 一個單一的返回值來消費該數據。

function add(x, y) {
  console.log("Hello");
  return x + y;
}
const x = add(4, 5); 

ES6介紹了 Iterator 迭代器 和 Generator 生成器，另一種 Pull 體系，調用 iterator.next() 的代碼是消費者，可從中拉取多個值。

在 Push 體系中，數據的生產者決定何時發送數據給消費者，消費者不會在接收數據之前意識到它將要接收這個數據。

Promise 是當今 JS 中最常見的 Push 體系，一個 Promise （數據的生產者）發送一個 resolved value （成功狀態的值）來執行一個回調（數據消費者）。但是不同于函數的地方的是：Promise 決定著何時數據才被推送至這個回調函數。

RxJS 引入了 Observable (可觀察對象)，一個全新的 Push 體系。一個可觀察對象是一個產生多值的生產者，當產生新數據的時候，會主動 "推送給" Observer (觀察者)。

Observable（可觀察對象）是基于推送（Push）運行時執行（lazy）的多值集合。

Observable 于 Promise 之間的差異：

Promise：只能返回單個值，不可取消，要么 resolve 要么 reject 并且只響應一次。

Observable：隨著時間的推移發出多個值，可以調用 unsubscribe() 取消訂閱，支持 map、filter、reduce 等操作符，延遲執行，當訂閱的時候才會開始執行，可以響應多次。

使用 RxJS 我們可以使用 npm 進行安裝（更多使用方法請參考 github）：

npm install rxjs 

需要注意的是，很多人認為 RxJS 中的所有操作都是異步的，但其實這個觀念是錯的。RxJS 的核心特性是它的異步處理能力，但它也是可以用來處理同步的行為。具體示例如下：

import { Observable } from "rxjs";
 
const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});
 
console.log("start");
observable.subscribe({
  next(x) { console.log(x); },
  error(err) { console.error(err); },
  complete() { console.log("done"); }
});
console.log("end");

以上代碼運行后，控制臺的輸出結果：

start
1
2
3
done
end

當然我們也可以用它處理異步行為：

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("start");
observable.subscribe({
  next(x) { console.log(x); },
  error(err) { console.error(err); },
  complete() { console.log("done"); }
});
console.log("end");

代碼運行后的輸出結果為：

start
1
2
3
end
4
done

RxJS 中提供了很多操作符 Operators，下篇文章我們將對 Operators 進行介紹，創建類操作符（Creation Operator）用于創建 Observable 對象。

官網列舉的一些創建類操作符如下：

ajax

bindCallback

bindNodeCallback

defer

empty

from

fromEvent

fromEventPattern

generate

interval

of

range

throwError

timer

iif

最后我們簡單的來看一下，如何使用 from 創建一個 Observable（關于操作符的介紹我們將在下篇文章進行詳細介紹，保持關注哦）。

from：可以把數組、Promise、以及 Iterable 轉化為 Observable。

//將數組轉換為 Observable：

import { from } from "rxjs";

const array = [10, 20, 30];
const result = from(array);

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

// Logs:
// 10
// 20
// 30

由于 RxJS 涉及到的概念和知識點比較寬泛和復雜，我們需要一步一步的去理解和掌握它，最終可以做到知其然亦知其所以然。接下來文章中會繼續介紹 RxJS 中涉及到知識點，關注此公眾號webinfoq不要錯過哦。