摘要：是一個基于可觀測數據流在異步編程應用中的庫。正如官網所說，是基于觀察者模式，迭代器模式和函數式編程。它具有時間與事件響應的概念。通知不再發(fā)送任何值。和通知可能只會在執(zhí)行期間發(fā)生一次，并且只會執(zhí)行其中的一個。

RxJS是一個基于可觀測數據流在異步編程應用中的庫。

ReactiveX is a combination of the best ideas from
the Observer pattern, the Iterator pattern, and functional programming

正如官網所說，RxJS是基于觀察者模式，迭代器模式和函數式編程。因此，首先要對這幾個模式有所理解

window.addEventListener("click", function(){
  console.log("click!");
})

JS的事件監(jiān)聽就是天生的觀察者模式。給window的click事件(被觀察者)綁定了一個listener(觀察者)，當事件發(fā)生，回調函數就會被觸發(fā)

迭代器模式，提供一種方法順序訪問一個聚合對象中的各種元素，而又不暴露該對象的內部表示。

ES6里的Iterator即可實現：

let arr = ["a", "b", "c"];
let iter = arr[Symbol.iterator]();

iter.next() // { value: "a", done: false }
iter.next() // { value: "b", done: false }
iter.next() // { value: "c", done: false }
iter.next() // { value: undefined, done: true }

反復調用迭代對象的next方法，即可順序訪問

提到函數式編程，就要提到聲明式編程和命令式編程
函數式編程是聲明式編程的體現

問題：將數組[1, 2, 3]的每個元素乘以2，然后計算總和。

命令式編程

const arr = [1, 2, 3];
let total = 0;

for(let i = 0; i < arr.length; i++) {
  total += arr[i] * 2;
}

聲明式編程

const arr = [1, 2, 3];
let total = arr.map(x => x * 2).reduce((total, value) => total + value)

聲明式的特點是專注于描述結果本身，不關注到底怎么到達結果。而命令式就是真正實現結果的步驟

聲明式編程把原始數據經過一系列轉換(map, reduce)，最后得到想要的數據

現在前端流行的MVC框架(Vue,React,Angular)，也都是提倡：編寫UI結構時使用聲明式編程，在編寫業(yè)務邏輯時使用命令式編程

RxJS里有兩個重要的概念需要我們理解:
Observable (可觀察對象)
Observer (觀察者)

var btn = document.getElementById("btn");
var handler = function() {
  console.log("click");
}
btn.addEventListener("click", handler)

上面這個例子里：
btn這個DOM元素的click事件就是一個Observable
handler這個函數就是一個Observer，當btn的click事件被觸發(fā)，就會調用該函數

改用RxJS編寫；

Rx.Observable.fromEvent(btn, "click")
.subscribe(() => console.log("click"));

fromEvent把一個event轉成了一個Observable，然后它就可以被訂閱subscribe了

Observable其實就是數據流stream
流是在時間流逝的過程中產生的一系列事件。它具有時間與事件響應的概念。

我們可以把一切輸入都當做數據流來處理，比如說：

用戶操作

網絡響應

定時器

Worker

當產生了一個流后，我們可以通過操作符(Operator)對這個流進行一系列加工操作，然后產生一個新的流

Rx.Observable.fromEvent(window, "click")
  .map(e => 1)
  .scan((total, now) => total + now)
  .subscribe(value => {
    console.log(value)
  })

map把流轉換成了一個每次產生1的新流，然后scan類似reduce，也會產生一個新流，最后這個流被訂閱。最終實現了:每次點擊累加1的效果

可以用一個效果圖來表示該過程：

也可以對若干個數據流進行組合：

例子：我們要實現下面這個效果：

Rx.Observable.fromEvent(document.querySelector("input[name=plus]"), "click")
  .mapTo(1)
  .merge(
    Rx.Observable.fromEvent(document.querySelector("input[name=minus]"), "click")
      .mapTo(-1)
  )
  .scan((total, now) => total + now)
  .subscribe(value => {
    document.querySelector("#counter").innerText = value;
  })

merge可以把兩個數據流整個在一起，效果可以參考如下：

剛才那個例子的數據流如下：

以RxJS的寫法，就是把按下加1當成一個數據流，把按下減1當成一個數據流，再通過merge把兩個數據流合并，最后通過scan操作符，把新流上的數據累加，這就是我們想要的計數器效果

有時候，我們的Observable送出的是一個新的Observable:

var click = Rx.Observable.fromEvent(document.body, "click");
var source = click.map(e => Rx.Observable.of(1, 2, 3));
source.subscribe(value => {
  console.log(value)
});

這里，console打印出來的是對象，而不是我們想要的1,2,3，這是因為map返回的Rx.Observable.of(1, 2, 3)本身也是個Observable

用圖表示如下：

click  : ------c------------c--------

        map(e => Rx.Observable.of(1,2,3))

source : ------o------------o--------
                            
                 (123)|       (123)|

因此，我們訂閱到的value值就是一個Observable對象，而不是普通數據1，2，3

我想要的其實不是Observable本身，而是屬于這個Observable里面的那些東西，現在這個情形就是Observable里面又有Observable，有兩層，可是我想要讓它變成一層就好，該怎么辦呢？

這就需要把Observable扁平化

const arr = [1, [2, 3], 4];

// 扁平化后：
const flatArr = [1, 2, 3, 4];

concatAll這個操作符就可以把Observable扁平化

var click = Rx.Observable.fromEvent(document.body, "click");
var source = click.map(e => Rx.Observable.of(1, 2, 3));
var example = source.concatAll();

example.subscribe(value => {
  console.log(value)
})

click  : ------c------------c--------

        map(e => Rx.Observable.of(1,2,3))

source : ------o------------o--------
                            
                 (123)|       (123)|

                   concatAll()

example: ------(123)--------(123)------------

flatMap操作符也可以實現同樣的作用，就是寫法有些不同：

var click = Rx.Observable.fromEvent(document.body, "click");
var source = click.flatMap(e => Rx.Observable.of(1, 2, 3));
source.subscribe(value => {
  console.log(value)
})

click  : ------c------------c--------

        flatMap(e => Rx.Observable.of(1,2,3))

source: ------(123)--------(123)------------

學完前面幾個操作符，我們就可以寫一個簡單的實例了

拖拽的原理是：

監(jiān)聽拖拽元素的mousedown

監(jiān)聽body的mousemove

監(jiān)聽body的mouseup

const mouseDown = Rx.Observable.fromEvent(dragDOM, "mousedown");
const mouseUp = Rx.Observable.fromEvent(body, "mouseup");
const mouseMove = Rx.Observable.fromEvent(body, "mousemove");

首先給出3個Observable，分別代表3種事件，我們希望mousedown的時候監(jiān)聽mousemove,然后mouseup時停止監(jiān)聽，于是RxJS可以這么寫：

const source = mouseDown
.map(event => mouseMove.takeUntil(mouseUp))

takeUntil操作符可以在某個條件符合時，發(fā)送complete事件

source: -------e--------------e-----
                              
                  --m-m-m-m|     -m--m-m--m-m|

從圖上可以看出，我們還需要把source扁平化，才能獲取所需數據。

完整代碼：

const dragDOM = document.getElementById("drag");
const body = document.body;

const mouseDown = Rx.Observable.fromEvent(dragDOM, "mousedown");
const mouseUp = Rx.Observable.fromEvent(body, "mouseup");
const mouseMove = Rx.Observable.fromEvent(body, "mousemove");

mouseDown
    .flatMap(event => mouseMove.takeUntil(mouseUp))
    .map(event => ({ x: event.clientX, y: event.clientY }))
    .subscribe(pos => {
        dragDOM.style.left = pos.x + "px";
        dragDOM.style.top = pos.y + "px";
    })

前面的例子，我們都在討論fromEvent轉換的Observable，其實還有很多種方法產生一個Observable，其中create也是一種常見的方法，可以用來創(chuàng)建自定義的Observable

var observable = Rx.Observable.create(function (observer) {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  setTimeout(() => {
    observer.next(4);
    observer.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");

控制臺執(zhí)行的結果：

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

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

"Next" 通知： 發(fā)送一個值，比如數字、字符串、對象，等等。
"Error" 通知： 發(fā)送一個 JavaScript 錯誤 或 異常。
"Complete" 通知： 不再發(fā)送任何值。
"Next" 通知是最重要，也是最常見的類型：它們表示傳遞給觀察者的實際數據。"Error" 和 "Complete" 通知可能只會在 Observable 執(zhí)行期間發(fā)生一次，并且只會執(zhí)行其中的一個。

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); // 如果捕獲到異常會發(fā)送一個錯誤
  }
});

Observer觀察者只是一組回調函數的集合，每個回調函數對應一種 Observable 發(fā)送的通知類型：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"),
};

Observer和Observable是通過subscribe方法建立聯系的

observable.subscribe(observer);

observer訂閱了Observable之后，還可以取消訂閱

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

unsubscribe陷阱：

let stream$ = new Rx.Observable.create((observer) => {
  let i = 0;
  let id = setInterval(() => {
    console.log("setInterval");
    observer.next(i++);
  },1000)
})

let subscription = stream$.subscribe((value) => {
  console.log("Value", value)
});

setTimeout(() => {
  subscription.unsubscribe();
}, 3000)

3秒后雖然取消了訂閱，但是開啟的setInterval定時器并不會自動清理，我們需要自己返回一個清理函數

let stream$ = new Rx.Observable.create((observer) => {
  let i = 0;
  let id = setInterval(() => {
    observer.next(i++);
  },1000)

  // 返回了一個清理函數
  return function(){
    clearInterval( id );
  }
})

let subscription = stream$.subscribe((value) => {
  console.log("Value", value)
});

setTimeout(() => {
  subscription.unsubscribe() // 在這我們調用了清理函數

}, 3000)

function sendRequest(search) {
  return Rx.Observable.ajax.getJSON(`http://deepred5.com/cors.php?search=${search}`)
    .map(response => response)
}

Rx.Observable.fromEvent(document.querySelector("input"), "keyup")
  .map(e => e.target.value)
  .flatMap(search => sendRequest(search))
  .subscribe(value => {
    console.log(value)
  })

用戶每次在input框每次進行輸入，均會觸發(fā)ajax請求，并且每個ajax返回的值都會被打印一遍

現在需要實現這樣一個功能：
希望用戶在300ms以內停止輸入，才發(fā)送請求(防抖)，并且console打印出來的值只要最近的一個ajax返回的

Rx.Observable.fromEvent(document.querySelector("input"), "keyup")
  .debounceTime(300)
  .map(e => e.target.value)
  .switchMap(search => sendRequest(search))
  .subscribe(value => {
    console.log(value)
  })

debounceTime表示經過n毫秒后，沒有流入新值，那么才將值轉入下一個環(huán)節(jié)
switchMap能取消上一個已無用的請求，只保留最后的請求結果流，這樣就確保處理展示的是最后的搜索的結果

可以看到，RxJS對異步的處理是非常優(yōu)秀的，對異步的結果能進行各種復雜的處理和篩選。

Redux的action都是同步的，所以默認情況下也只能處理同步數據流。

為了生成異步action，處理異步數據流，有許多不同的解決方案，例如 redux-thunk、redux-promise、redux-saga 等等。

以redux-thunk舉例：

調用一個異步API，首先要先定義三個同步action構造函數，分別表示

請求開始

請求成功

請求失敗

然后再定義一個異步action構造函數，該函數不再是返回普通的對象，而是返回一個函數，在這個函數里，進行ajax異步操作，然后根據返回的成功和失敗，分別調用前面定義的同步action

actions.js

export const FETCH_STARTED = "WEATHER/FETCH_STARTED";
export const FETCH_SUCCESS = "WEATHER/FETCH_SUCCESS";
export const FETCH_FAILURE = "WEATHER/FETCH_FAILURE";

// 普通action構造函數，返回普通對象
export const fetchWeatherStarted = () => ({
  type: FETCH_STARTED
});

export const fetchWeatherSuccess = (result) => ({
  type: FETCH_SUCCESS,
  result
})

export const fetchWeatherFailure = (error) => ({
  type: FETCH_FAILURE,
  error
})

// 異步action構造函數，返回一個函數
export const fetchWeather = (cityCode) => {
  return (dispatch) => {
    const apiUrl = `/data/cityinfo/${cityCode}.html`;

    dispatch(fetchWeatherStarted())

    return fetch(apiUrl).then((response) => {
      if (response.status !== 200) {
        throw new Error("Fail to get response with status " + response.status);
      }

      response.json().then((responseJson) => {
        dispatch(fetchWeatherSuccess(responseJson.weatherinfo));
      }).catch((error) => {
        dispatch(fetchWeatherFailure(error));
      });
    }).catch((error) => {
      dispatch(fetchWeatherFailure(error));
    })
  };
}

現在如果想要異步請求，只要:

// fetchWeather是個異步action構造函數
dispatch(fetchWeather("23333"));

我們再來看看redux-observable:

調用一個異步API，不再需要定義一個異步action構造函數，所有的action構造函數都只是返回普通的對象

那么ajax請求在哪里發(fā)送？

答案是在Epic進行異步操作

Epic是redux-observable的核心原語。
它是一個函數，接收 actions 流作為參數并且返回 actions 流。 Actions 入, actions 出.

export const FETCH_STARTED = "WEATHER/FETCH_STARTED";
export const FETCH_SUCCESS = "WEATHER/FETCH_SUCCESS";
export const FETCH_FAILURE = "WEATHER/FETCH_FAILURE";

export const fetchWeather = cityCode => ({ type: FETCH_STARTED, cityCode });
export const fetchWeatherSuccess = result => (
  { type: FETCH_SUCCESS, result };
);
export const fetchWeatherFailure = (error) => (
  {
    type: FETCH_FAILURE,
    error
  }
)

export const fetchWeatherEpic = action$ =>
  action$.ofType(FETCH_STARTED)
    .mergeMap(action =>
      ajax.getJSON(`/data/cityinfo/${action.cityCode}.html`)
        .map(response => fetchWeatherSuccess(response.weatherinfo))
        // 這個處理異常的action必須使用Observable.of方法轉為一個observable
        .catch(error => Observable.of(fetchWeatherFailure(error)))
    );

現在如果想要異步請求，只要:

// fetchWeather只是個普通的action構造函數
dispatch(fetchWeather("23333"));

相較于thunk中間件，使用redux-observable來處理異步action，有以下優(yōu)點：

不需要修改action構造函數，返回的仍然是普通對象

epics中間件會將action封裝成Observable對象，可以使用RxJs的相應api來控制異步流程，它就像一個擁有許多高級功能的Promise，現在我們在Redux中也可以得到它的好處。

原生JS傳統(tǒng)解決異步的方式：callback、Generator、Promise、async/await

RxJS解決的是數據流的問題，它可以讓批量數據處理起來更方便

可以想象的一些使用場景：

多個服務端實時消息流，通過RxJS進行高階處理，最后到 view 層就是很清晰的一個Observable，但是view層本身處理用戶事件依然可以沿用原有的范式。

爬蟲抓取，每次對一個網站的前5頁做平行請求，每個請求如果失敗就重試，重試3次之后再放棄。

可以看出，這種需要對流進行復雜操作的場景更加適合RxJS

公司內部目前的大部分系統(tǒng)，前端就可能不太適合用RxJS，因為大部分是后臺CRUD系統(tǒng)，整體性、實時性的要求都不高，并且也沒有特別復雜的數據流操作

我們推薦在適合RxJS的地方用RxJS，但是不強求RxJS for everything。RxJS給了我們另一種思考和解決問題的方式，但這不一定是必要的

構建流式應用—RxJS詳解

希望是最淺顯易懂的RxJS教學

RxJS入門指引和初步應用

30天精通RxJS系列