摘要：更加優(yōu)雅的風(fēng)格按照上面的教程，我們在中獲取到了數(shù)據(jù)發(fā)送異步請求并拿到了最新一次的返回值。之后，再通過，在監(jiān)聽的回調(diào)中將返回值拼接成并插入。

本文是一篇 RxJS 實戰(zhàn)教程，利用 RxJS 和 github API 來一步步做一個 github 小應(yīng)用。因此，文章的重點是解釋 RxJS 的使用，而涉及的 ES6語法、webpack 等知識點不予講解。

本例的所有代碼在 github 倉庫：rxjs-example

首先要注意的是，目前在 github 上有兩個主流 RxJS，它們代表不同的版本：

ReactiveX - rxjs RxJS 5 beta 版

Reactive-Extensions - RxJS RxJS 4.x 穩(wěn)定版

這兩個版本的安裝和引用稍有不同：

# 安裝 4.x 穩(wěn)定版
$ npm install rx --save
# 安裝 5 beta 版
$ npm install rxjs --save

// 4.x 穩(wěn)定版
import Rx from "rx";
// 5 beta 版
import Rx from "rxjs/Rx";

除此以外，它們的語法也稍有不同，比如在 5 beta 版里，subscribe時可以代入一個對象作為參數(shù)，也可以代入回調(diào)函數(shù)作為參數(shù)，而 4.x 版則只支持以回調(diào)函數(shù)為參數(shù)的情況：

// 5 beta
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);

// 5 和 4.x 都支持：
Observable.subscribe(x => console.log(x), (err) => console.log(err), () => console.log("completed"));

其他更多語法不同可以參考：

4.x 穩(wěn)定版 Document

5 beta 版 Document

從 4 到 5 的遷移

如上所說，我們要利用 RxJS 和 github API 來一步步做一個 github 小應(yīng)用。首先完成其基本功能，即通過一個 input 輸入文字，并實時根據(jù) input 內(nèi)值的變化去發(fā)送異步請求，調(diào)用 github API 進行搜索。如圖所示（線上 Demo）：

通過RxJS，在輸入過程中實時進行異步搜索：

hover到 avator 上之后異步獲取用戶信息

安裝 webpack 配置編譯環(huán)境，并使用 ES6 語法。安裝如下依賴，并配置好 webpack：

webpack

webpack-dev-server

babel-loader

babel-preset-es2015

html-webpack-plugin

css-loader / postcss 及其他

jquery

rx（4.x 版本）

通過webpack-dev-server，我們將會啟動一個 8080 端口的服務(wù)器，使得我們編譯好的資源可以在localhost:8080/webpack-dev-server訪問到。

在index.html中編寫一個input，我們將在index.js中，通過 RxJS 的 Observable 監(jiān)聽input的keyup事件。可以使用fromEvent來創(chuàng)建一個基于 DOM 事件的流，并通過map和filter進一步處理。

// src/js/index.js
import Rx from "rx";

$(() => {
  const $input = $(".search");
  // 通過 input 的 keyup 事件來創(chuàng)建流
  const observable = Rx.Observable.fromEvent($input, "keyup")
      // 并獲取每次 keyup 時搜索框的值，篩選出合法值
      .map(() => $input.val().trim())
    .filter((text) => !!text)
    // 利用 do 可以做一些不影響流的事件，比如這里打印出 input 的值
    .do((value) => console.log(value));
  // 開啟監(jiān)聽
  observable.subscribe();
});

去 input 里隨便打打字，可以看到我們已經(jīng)成功監(jiān)聽了keyup事件，并在每次keyup時在 console 里輸出 input 當(dāng)前的值。

監(jiān)聽了 input 事件，我們就能夠在每次keyup時拿到 value，那么就可以通過它來異步獲取數(shù)據(jù)。將整個過程拆分一下：

用戶在 input 里輸入任意內(nèi)容

觸發(fā)keyup事件，獲取到當(dāng)前 value

將 value 代入到一個異步方法里，通過接口獲取數(shù)據(jù)

利用返回數(shù)據(jù)渲染 DOM

也就是說，我們要把原有的 Observable 中每個事件返回的 value 進行異步處理，并使其返回一個新的 Observable?？梢赃@么處理：

讓每個 value 返回一個 Observable

通過flatMap將所有的 Observable 扁平化，成為一個新的 Observable

圖解flatMap：

而既然需要異步獲取數(shù)據(jù)，那么在上面的第一步時，可以通過fromPromise來創(chuàng)建一個 Observable：

// src/js/helper.js
const SEARCH_REPOS = "https://api.github.com/search/repositories?sort=stars&order=desc&q=";

// 創(chuàng)建一個 ajax 的 promise
const getReposPromise = (query) => {
  return $.ajax({
      type: "GET",
    url: `${SEARCH_REPOS}${query}`,
  }).promise();
};
// 通過 fromPromise 創(chuàng)建一個 Observable
export const getRepos = (query) => {
  const promise = getReposPromise(query);
  return Rx.Observable.fromPromise(promise);
};

// src/js/index.js
import {getRepos} from "./helper";

// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
      .map(() => $input.val())
    .filter((text) => !!text)
    .do((value) => console.log(value))
    // 調(diào)用 getRepos 方法將返回一個 Observable
    // flatMap 則將所有 Observable 合并，轉(zhuǎn)為一個 Observable
    .flatMap(getRepos);
// ...

這樣，每一次keyup的時候，都會根據(jù)此時 input 的 value 去異步獲取數(shù)據(jù)。但這樣做有幾個問題：

不斷打字時會連續(xù)不斷觸發(fā)異步請求，占用資源影響體驗

如果相鄰的keyup事件觸發(fā)時 input 的值一樣，也就是說按下了不改變 value 的按鍵（比如方向鍵），會重復(fù)觸發(fā)一樣的異步事件

發(fā)出多個異步事件之后，每個事件所耗費的時間不一定相同。如果前一個異步所用時間較后一個長，那么當(dāng)它最終返回結(jié)果時，有可能把后面的異步率先返回的結(jié)果覆蓋

所以接下來我們就處理這幾個問題。

針對上面的問題，一步一步進行優(yōu)化。

不斷打字時會連續(xù)不斷觸發(fā)異步請求，占用資源影響體驗

也就是說，當(dāng)用戶在連續(xù)打字時，我們不應(yīng)該繼續(xù)進行之后的事件處理，而如果打字中斷，或者說兩次keyup事件的時間間隔足夠長時，才應(yīng)該發(fā)送異步請求。針對這點，可以使用 RxJS 的debounce方法：

如圖所示，在一段時間內(nèi)事件被不斷觸發(fā)時，不會被之后的操作所處理；只有超過指定時間間隔的事件才會留下來：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    // 若 400ms 內(nèi)連續(xù)觸發(fā) keyup 事件，則不會繼續(xù)往下處理
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .do((value) => console.log(value))
    .flatMap(getRepos);
// ...

如果相鄰的keyup事件觸發(fā)時 input 的值一樣，也就是說按下了不改變 value 的按鍵（比如方向鍵），會重復(fù)觸發(fā)一樣的異步事件

也就是說，對于任意相鄰的事件，如果它們的返回值一樣，則只要取一個（重復(fù)事件中的第一個）就好了?？梢岳?b>distinctUntilChanged方法：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    // 只取不一樣的值進行異步
    .distinctUntilChanged()
    .do((value) => console.log(value))
    .flatMap(getRepos);
// ...

發(fā)出多個異步事件之后，每個事件所耗費的時間不一定相同。如果前一個異步所用時間較后一個長，那么當(dāng)它最終返回結(jié)果時，有可能把后面的異步率先返回的結(jié)果覆蓋

這個蛋疼的問題我相信大家很可能遇見過。在發(fā)送多個異步請求時，因為所用時長不一定，無法保障異步返回的先后順序，所以，有時候可能早請求的異步的結(jié)果會覆蓋后來請求的異步結(jié)果。

而這種情況的處理方式就是，在連續(xù)發(fā)出多個異步的時候，既然我們期待的是最后一個異步返回的結(jié)果，那么就可以把之前的異步取消掉，不 care 其返回了什么。因此，我們可以使用flatMapLatest API（類似于 RxJava 中的switchMap API，同時在 RxJS 5.0 中也已經(jīng)改名為switchMap）

通過flatMapLatest，當(dāng) Observable 觸發(fā)某個事件，返回新的 Observable 時，將取消之前觸發(fā)的事件，并且不再關(guān)心返回結(jié)果的處理，只監(jiān)視當(dāng)前這一個。也就是說，發(fā)送多個請求時，不關(guān)心之前請求的處理，只處理最后一次的請求：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .distinctUntilChanged()
    .do((value) => console.log(value))
    // 僅處理最后一次的異步
    .flatMapLatest(getRepos);
// ...

至此，我們對 input keyup以及異步獲取數(shù)據(jù)的整個事件流處理完畢，并進行了一定的優(yōu)化，避免了過多的請求、異步返回結(jié)果錯亂等問題。但創(chuàng)建了一個流之后也有對其進行監(jiān)聽：

// src/js/index.js
// ...
const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .distinctUntilChanged()
    .do((value) => console.log(value))
    .flatMapLatest(getRepos);
// 第一個回調(diào)中的 data 代表異步的返回值
observable.subscribe((data) => {
  // 在 showNewResults 方法中使用返回值渲染 DOM
  showNewResults(data);
}, (err) => {
  console.log(err);
}, () => {
  console.log("completed");
});

// 異步返回的結(jié)果是個 Array，代表搜索到的各個倉庫 item
// 遍歷所有 item，轉(zhuǎn)化為 jQuery 對象，最后插入到 content_container 中
const showNewResults = (items) => {
  const repos = items.map((item, i) => {
    return reposTemplate(item);
  }).join("");
  $(".content_container").html(repos);
};

這樣，一個通過 RxJS 監(jiān)聽事件的流已經(jīng)完全建立完畢了。整個過程使用圖像來表示則如下：

而如果我們不使用 RxJS，用傳統(tǒng)方式監(jiān)聽 input 的話：

// src/js/index.js
import {getRepos} from "./helper";

$(() => {
  const $input = $(".search");
  const interval = 400;
  var previousValue = null;
  var fetching = false;
  var lastKeyUp = Date.now() - interval;
  $input.on("keyup", (e) => {
    const nextValue = $input.val();
    if (!nextValue) {
      return;
    }
    if (Date.now() - lastKeyUp <= interval) {
      return;
    }
    lastKeyUp = Date.now();
    if (nextValue === previousValue) {
      return;
    }
    previousValue = nextValue;
    if (!fetching) {
      fetching = true;
      getRepos(nextValue).then((data) => {
          fetching = false;
        showNewResults(data);
      });
    }
  });
});

挺復(fù)雜了吧？而且即便如此，這樣的處理還是不夠到位。上面僅僅是通過fetching變量來判斷是否正在異步，如果正在異步，則不進行新的異步；而我們更希望的是能夠取消舊的異步，只處理新的異步請求。

按照上面的教程，我們在 Observable 中獲取到了數(shù)據(jù)、發(fā)送異步請求并拿到了最新一次的返回值。之后，再通過subscribe，在監(jiān)聽的回調(diào)中將返回值拼接成 HTML 并插入 DOM。

但是有一個問題：小應(yīng)用的另一個功能是，當(dāng)鼠標hover到頭像上時，異步獲取并展現(xiàn)用戶的信息?？墒怯脩纛^像是在subscribe回調(diào)中動態(tài)插入的，又該如何創(chuàng)建事件流呢？當(dāng)然了，可以在每次插入 DOM 之后在利用fromEvent創(chuàng)建一個基于hover的事件流，但那樣總是不太好的，寫出來的代碼也不夠 Rx。或許我們就不應(yīng)該在.flatMapLatest(getRepos)之后中斷流的傳遞？但那樣的話，又該如何把異步的返回值插入 DOM 呢？

針對這種情況，我們可以使用 RxJS 的do方法：

你想在do的回調(diào)內(nèi)做什么都可以，它不會影響到流內(nèi)的事件；除此以外，還可以拿到流中各個事件的返回值：

var observable = Rx.Observable.from([0, 1, 2])
    .do((x) => console.log(x))
    .map((x) => x + 1);
observable.subscribe((x) => {
  console.log(x);
});

所以，我們可以利用do來完成 DOM 的渲染：

// src/js/index.js
// ...
// $conatiner 是裝載搜索結(jié)果的容器 div
const $conatiner = $(".content_container");

const observable = Rx.Observable.fromEvent($input, "keyup")
    .debounce(400)
      .map(() => $input.val())
    .filter((text) => !!text)
    .distinctUntilChanged()
    .do((value) => console.log(value))
    .flatMapLatest(getRepos)
    // 首先把之前的搜索結(jié)果清空
    .do((results) => $conatiner.html(""))
    // 利用 Rx.Observable.from 將異步的結(jié)果轉(zhuǎn)化為 Observable，并通過 flatMap 合并到原有的流中。此時流中的每個元素是 results 中的每個 item
    .flatMap((results) => Rx.Observable.from(results))
    // 將各 item 轉(zhuǎn)化為 jQuery 對象
    .map((repos) => $(reposTemplate(repos)))
    // 最后把每個 jQuery 對象依次加到容器里
    .do(($repos) => {
      $conatiner.append($repos);
    });

// 在 subscribe 中實際上什么都不用做，就能達到之前的效果
observable.subscribe(() => {
  console.log("success");
}, (err) => {
  console.log(err);
}, () => {
  console.log("completed");
});

簡直完美！現(xiàn)在我們這個observable在最后通過map，依次返回了一個 jQuery 對象。那么之后如果要對頭像添加hover的監(jiān)聽，則可以在這個流的基礎(chǔ)上繼續(xù)進行。

我們接下來針對用戶頭像的hover事件創(chuàng)建一個流。用戶的詳細資料是異步加載的，而hover到頭像上時彈出 modal。如果是第一個hover，則 modal 里只有一個 loading 的圖標，并且異步獲取數(shù)據(jù)，之后將返回的數(shù)據(jù)插入到 modal 里；而如果已經(jīng)拿到并插入好了數(shù)據(jù)，則不再有異步請求，直接展示：

沒有數(shù)據(jù)時展示 loading，同時異步獲取數(shù)據(jù)

異步返回后插入數(shù)據(jù)。且如果已經(jīng)有了數(shù)據(jù)則直接展示

先不管上一個流，我們先創(chuàng)建一個新的事件流：

// src/js/index.js
// ...
const initialUserInfoSteam = () => {
  const $avator = $(".user_header");
  // 通過頭像 $avator 的 hover 事件來創(chuàng)建流
  const avatorMouseover = Rx.Observable.fromEvent($avator, "mouseover")
    // 500ms 內(nèi)重復(fù)觸發(fā)事件則會被忽略
    .debounce(500)
    // 只有當(dāng)滿足了下列條件的流才會繼續(xù)執(zhí)行，否則將中斷
    .takeWhile((e) => {
      // 異步獲取的用戶信息被新建到 DOM 里，該 DOM 最外層是 infos_container
      // 因此，如果已經(jīng)有了 infos_container，則可以認為我們已經(jīng)異步獲取過數(shù)據(jù)了，此時 takeWhile 將返回 false，流將會中斷
      const $infosWrapper = $(e.target).parent().find(".user_infos_wrapper");
      return $infosWrapper.find(".infos_container").length === 0;
    })
    .map((e) => {
      const $infosWrapper = $(e.target).parent().find(".user_infos_wrapper");
      return {
        conatiner: $infosWrapper,
        url: $(e.target).attr("data-api")
      }
    })
    .filter((data) => !!data.url)
    // getUser 來異步獲取用戶信息
    .flatMapLatest(getUser)
    .do((result) => {
      // 將用戶信息組建成為 DOM 元素，并插入到頁面中。在這之后，該用戶對應(yīng)的 DOM 里就會擁有 infos_container 這個 div，所以 takeWhile 會返回 false。也就是說，之后再 hover 上去，流也不會被觸發(fā)了
      const {data, conatiner} = result;
      showUserInfo(conatiner, data);
    });

  avatorMouseover.subscribe((result) => {
      console.log("fetch user info succeed");
  }, (err) => {
    console.log(err);
  }, () => {
    console.log("completed");
  });
};

上面的代碼中有一個 API 需要講解：takeWhile

由圖可知，當(dāng)takeWhile中的回調(diào)返回true時，流可以正常進行；而一旦返回false，則之后的事件不會再發(fā)生，流將直接終止：

var source = Rx.Observable.range(1, 5)
    .takeWhile(function (x) { return x < 3; });

var subscription = source.subscribe(
    function (x) { console.log("Next: " + x); },
    function (err) { console.log("Error: " + err); },
    function () { console.log("Completed"); });
// Next: 0
// Next: 1
// Next: 2
// Completed

創(chuàng)建好針對hover的事件流，我們可以把它和上一個事件流結(jié)合起來：

// src/js/index.js
// ...
const initialUserInfoSteam = ($repos) => {
  const $avator = $repos.find(".user_header");
  // ...
}

const observable = Rx.Observable.fromEvent($input, "keyup")
    // ...
    .do(($repos) => {
      $conatiner.append($repos);
      initialUserInfoSteam($repos);
    });
// ...

現(xiàn)在這樣就已經(jīng)可以使用了，但依舊不夠好。目前總共有兩個流：監(jiān)聽 input keyup的流和監(jiān)聽mouseover的流。但是，因為用戶頭像是動態(tài)插入的 ，所以我們必須在$conatiner.append($repos);之后才能創(chuàng)建并監(jiān)聽mouseover。不過鑒于我們已經(jīng)在最后的do方法里插入了獲取的數(shù)據(jù)，所以可以試著把兩個流合并到一起：

// src/js/index.js
// ...
const initialUserInfoSteam = ($repos) => {
  const $avator = $repos.find(".user_header");
  const avatorMouseover = Rx.Observable.fromEvent($avator, "mouseover")
  // ... 流的處理跟之前的一樣
  // 但我們不再需要 subscribe 它，而是返回這個 Observable
  return avatorMouseover;
};

const observable = Rx.Observable.fromEvent($input, "keyup")
    // ...
    .do(($repos) => {
      $conatiner.append($repos);
      // 不再在 do 里面創(chuàng)建新的流并監(jiān)聽
      // initialUserInfoSteam($repos);
    })
    // 相反，我們繼續(xù)這個流的傳遞，只是通過 flatMap 將原來的流變成了監(jiān)聽 mouseover 的流
    .flatMap(($repos) => {
      return initialUserInfoSteam($repos);
    });
// ...

DONE ！

栗子中使用到的 RxJS API：

from 通過一個可迭代對象來創(chuàng)建流

fromEvent 通過 DOM 事件來創(chuàng)建流

debounce 如果在一定時間內(nèi)流中的某個事件不斷被觸發(fā)，則不會進行之后的事件操作

map 遍歷流中所有事件，返回新的流

filter 篩選流中所有事件，返回新的流

flatMap 對各個事件返回的值進行處理并返回 Observable，然后將所有的 Observable 扁平化，成為一個新的 Observable

flatMapLatest 對各個事件返回的值進行處理并返回 Observable，然后將所有的 Observable 扁平化，成為一個新的 Observable。但只會獲取最后一次返回的 Observable，其他的返回結(jié)果不予處理

distinctUntilChanged 流中如果相鄰事件的結(jié)果一樣，則僅篩選出一個（剔除重復(fù)值）

do 可以依次拿到流上每個事件的返回值，利用其做一些無關(guān)流傳遞的事情

takeWhile 給予流一個判斷，只有當(dāng)takeWhile中的回調(diào)返回true時，流才會繼續(xù)執(zhí)行；否則將中斷之后的事件

What does RxJS observable debounce do

How do I work with jQuery and RxJS

Introduction of observable operators

文章源碼 - rxjs-example