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

RxJS是一個(gè)基于可觀測(cè)數(shù)據(jù)流在異步編程應(yīng)用中的庫。

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

正如官網(wǎng)所說，RxJS是基于觀察者模式，迭代器模式和函數(shù)式編程。因此，首先要對(duì)這幾個(gè)模式有所理解

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

JS的事件監(jiān)聽就是天生的觀察者模式。給window的click事件(被觀察者)綁定了一個(gè)listener(觀察者)，當(dāng)事件發(fā)生，回調(diào)函數(shù)就會(huì)被觸發(fā)

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

ES6里的Iterator即可實(shí)現(xiàn)：

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 }

反復(fù)調(diào)用迭代對(duì)象的next方法，即可順序訪問

提到函數(shù)式編程，就要提到聲明式編程和命令式編程
函數(shù)式編程是聲明式編程的體現(xiàn)

問題：將數(shù)組[1, 2, 3]的每個(gè)元素乘以2，然后計(jì)算總和。

命令式編程

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)

聲明式的特點(diǎn)是專注于描述結(jié)果本身，不關(guān)注到底怎么到達(dá)結(jié)果。而命令式就是真正實(shí)現(xiàn)結(jié)果的步驟

聲明式編程把原始數(shù)據(jù)經(jīng)過一系列轉(zhuǎn)換(map, reduce)，最后得到想要的數(shù)據(jù)

現(xiàn)在前端流行的MVC框架(Vue,React,Angular)，也都是提倡：編寫UI結(jié)構(gòu)時(shí)使用聲明式編程，在編寫業(yè)務(wù)邏輯時(shí)使用命令式編程

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

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

上面這個(gè)例子里：
btn這個(gè)DOM元素的click事件就是一個(gè)Observable
handler這個(gè)函數(shù)就是一個(gè)Observer，當(dāng)btn的click事件被觸發(fā)，就會(huì)調(diào)用該函數(shù)

改用RxJS編寫；

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

fromEvent把一個(gè)event轉(zhuǎn)成了一個(gè)Observable，然后它就可以被訂閱subscribe了

Observable其實(shí)就是數(shù)據(jù)流stream
流是在時(shí)間流逝的過程中產(chǎn)生的一系列事件。它具有時(shí)間與事件響應(yīng)的概念。

我們可以把一切輸入都當(dāng)做數(shù)據(jù)流來處理，比如說：

用戶操作

網(wǎng)絡(luò)響應(yīng)

定時(shí)器

Worker

當(dāng)產(chǎn)生了一個(gè)流后，我們可以通過操作符(Operator)對(duì)這個(gè)流進(jìn)行一系列加工操作，然后產(chǎn)生一個(gè)新的流

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

map把流轉(zhuǎn)換成了一個(gè)每次產(chǎn)生1的新流，然后scan類似reduce，也會(huì)產(chǎn)生一個(gè)新流，最后這個(gè)流被訂閱。最終實(shí)現(xiàn)了:每次點(diǎn)擊累加1的效果

可以用一個(gè)效果圖來表示該過程：

也可以對(duì)若干個(gè)數(shù)據(jù)流進(jìn)行組合：

例子：我們要實(shí)現(xiàn)下面這個(gè)效果：

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可以把兩個(gè)數(shù)據(jù)流整個(gè)在一起，效果可以參考如下：

剛才那個(gè)例子的數(shù)據(jù)流如下：

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

有時(shí)候，我們的Observable送出的是一個(gè)新的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打印出來的是對(duì)象，而不是我們想要的1,2,3，這是因?yàn)?b>map返回的Rx.Observable.of(1, 2, 3)本身也是個(gè)Observable

用圖表示如下：

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

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

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

因此，我們訂閱到的value值就是一個(gè)Observable對(duì)象，而不是普通數(shù)據(jù)1，2，3

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

這就需要把Observable扁平化

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

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

concatAll這個(gè)操作符就可以把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操作符也可以實(shí)現(xiàn)同樣的作用，就是寫法有些不同：

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

學(xué)完前面幾個(gè)操作符，我們就可以寫一個(gè)簡(jiǎn)單的實(shí)例了

拖拽的原理是：

監(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個(gè)Observable，分別代表3種事件，我們希望mousedown的時(shí)候監(jiān)聽mousemove,然后mouseup時(shí)停止監(jiān)聽，于是RxJS可以這么寫：

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

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

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

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

完整代碼：

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轉(zhuǎn)換的Observable，其實(shí)還有很多種方法產(chǎn)生一個(gè)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");

控制臺(tái)執(zhí)行的結(jié)果：

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

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

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

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

Observer觀察者只是一組回調(diào)函數(shù)的集合，每個(gè)回調(diào)函數(shù)對(duì)應(yīng)一種 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方法建立聯(lián)系的

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定時(shí)器并不會(huì)自動(dòng)清理，我們需要自己返回一個(gè)清理函數(shù)

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

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

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

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

}, 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框每次進(jìn)行輸入，均會(huì)觸發(fā)ajax請(qǐng)求，并且每個(gè)ajax返回的值都會(huì)被打印一遍

現(xiàn)在需要實(shí)現(xiàn)這樣一個(gè)功能：
希望用戶在300ms以內(nèi)停止輸入，才發(fā)送請(qǐng)求(防抖)，并且console打印出來的值只要最近的一個(gè)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表示經(jīng)過n毫秒后，沒有流入新值，那么才將值轉(zhuǎn)入下一個(gè)環(huán)節(jié)
switchMap能取消上一個(gè)已無用的請(qǐng)求，只保留最后的請(qǐng)求結(jié)果流，這樣就確保處理展示的是最后的搜索的結(jié)果

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

Redux的action都是同步的，所以默認(rèn)情況下也只能處理同步數(shù)據(jù)流。

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

以redux-thunk舉例：

調(diào)用一個(gè)異步API，首先要先定義三個(gè)同步action構(gòu)造函數(shù)，分別表示

請(qǐng)求開始

請(qǐng)求成功

請(qǐng)求失敗

然后再定義一個(gè)異步action構(gòu)造函數(shù)，該函數(shù)不再是返回普通的對(duì)象，而是返回一個(gè)函數(shù)，在這個(gè)函數(shù)里，進(jìn)行ajax異步操作，然后根據(jù)返回的成功和失敗，分別調(diào)用前面定義的同步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構(gòu)造函數(shù)，返回普通對(duì)象
export const fetchWeatherStarted = () => ({
  type: FETCH_STARTED
});

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

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

// 異步action構(gòu)造函數(shù)，返回一個(gè)函數(shù)
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));
    })
  };
}

現(xiàn)在如果想要異步請(qǐng)求，只要:

// fetchWeather是個(gè)異步action構(gòu)造函數(shù)
dispatch(fetchWeather("23333"));

我們?cè)賮砜纯?b>redux-observable:

調(diào)用一個(gè)異步API，不再需要定義一個(gè)異步action構(gòu)造函數(shù)，所有的action構(gòu)造函數(shù)都只是返回普通的對(duì)象

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

答案是在Epic進(jìn)行異步操作

Epic是redux-observable的核心原語。
它是一個(gè)函數(shù)，接收 actions 流作為參數(shù)并且返回 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))
        // 這個(gè)處理異常的action必須使用Observable.of方法轉(zhuǎn)為一個(gè)observable
        .catch(error => Observable.of(fetchWeatherFailure(error)))
    );

現(xiàn)在如果想要異步請(qǐng)求，只要:

// fetchWeather只是個(gè)普通的action構(gòu)造函數(shù)
dispatch(fetchWeather("23333"));

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

不需要修改action構(gòu)造函數(shù)，返回的仍然是普通對(duì)象

epics中間件會(huì)將action封裝成Observable對(duì)象，可以使用RxJs的相應(yīng)api來控制異步流程，它就像一個(gè)擁有許多高級(jí)功能的Promise，現(xiàn)在我們?cè)赗edux中也可以得到它的好處。

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

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

可以想象的一些使用場(chǎng)景：

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

爬蟲抓取，每次對(duì)一個(gè)網(wǎng)站的前5頁做平行請(qǐng)求，每個(gè)請(qǐng)求如果失敗就重試，重試3次之后再放棄。

可以看出，這種需要對(duì)流進(jìn)行復(fù)雜操作的場(chǎng)景更加適合RxJS

公司內(nèi)部目前的大部分系統(tǒng)，前端就可能不太適合用RxJS，因?yàn)榇蟛糠质呛笈_(tái)CRUD系統(tǒng)，整體性、實(shí)時(shí)性的要求都不高，并且也沒有特別復(fù)雜的數(shù)據(jù)流操作

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

構(gòu)建流式應(yīng)用—RxJS詳解

希望是最淺顯易懂的RxJS教學(xué)

RxJS入門指引和初步應(yīng)用

30天精通RxJS系列