摘要：更加優雅的風格按照上面的教程，我們在中獲取到了數據發送異步請求并拿到了最新一次的返回值。之后，再通過，在監聽的回調中將返回值拼接成并插入。

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

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

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

ReactiveX - rxjs RxJS 5 beta 版

Reactive-Extensions - RxJS RxJS 4.x 穩定版

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

# 安裝 4.x 穩定版
$ npm install rx --save
# 安裝 5 beta 版
$ npm install rxjs --save

// 4.x 穩定版
import Rx from "rx";
// 5 beta 版
import Rx from "rxjs/Rx";

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

// 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 穩定版 Document

5 beta 版 Document

從 4 到 5 的遷移

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

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

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

安裝 webpack 配置編譯環境，并使用 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 端口的服務器，使得我們編譯好的資源可以在localhost:8080/webpack-dev-server訪問到。

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

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

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

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

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

用戶在 input 里輸入任意內容

觸發keyup事件，獲取到當前 value

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

利用返回數據渲染 DOM

也就是說，我們要把原有的 Observable 中每個事件返回的 value 進行異步處理，并使其返回一個新的 Observable。可以這么處理：

讓每個 value 返回一個 Observable

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

圖解flatMap：

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

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

// 創建一個 ajax 的 promise
const getReposPromise = (query) => {
  return $.ajax({
      type: "GET",
    url: `${SEARCH_REPOS}${query}`,
  }).promise();
};
// 通過 fromPromise 創建一個 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))
    // 調用 getRepos 方法將返回一個 Observable
    // flatMap 則將所有 Observable 合并，轉為一個 Observable
    .flatMap(getRepos);
// ...

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

不斷打字時會連續不斷觸發異步請求，占用資源影響體驗

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

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

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

針對上面的問題，一步一步進行優化。

不斷打字時會連續不斷觸發異步請求，占用資源影響體驗

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

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

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

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

也就是說，對于任意相鄰的事件，如果它們的返回值一樣，則只要取一個（重復事件中的第一個）就好了。可以利用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);
// ...

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

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

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

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

// 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以及異步獲取數據的整個事件流處理完畢，并進行了一定的優化，避免了過多的請求、異步返回結果錯亂等問題。但創建了一個流之后也有對其進行監聽：

// 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);
// 第一個回調中的 data 代表異步的返回值
observable.subscribe((data) => {
  // 在 showNewResults 方法中使用返回值渲染 DOM
  showNewResults(data);
}, (err) => {
  console.log(err);
}, () => {
  console.log("completed");
});

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

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

而如果我們不使用 RxJS，用傳統方式監聽 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);
      });
    }
  });
});

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

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

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

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

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

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 是裝載搜索結果的容器 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)
    // 首先把之前的搜索結果清空
    .do((results) => $conatiner.html(""))
    // 利用 Rx.Observable.from 將異步的結果轉化為 Observable，并通過 flatMap 合并到原有的流中。此時流中的每個元素是 results 中的每個 item
    .flatMap((results) => Rx.Observable.from(results))
    // 將各 item 轉化為 jQuery 對象
    .map((repos) => $(reposTemplate(repos)))
    // 最后把每個 jQuery 對象依次加到容器里
    .do(($repos) => {
      $conatiner.append($repos);
    });

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

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

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

沒有數據時展示 loading，同時異步獲取數據

異步返回后插入數據。且如果已經有了數據則直接展示

先不管上一個流，我們先創建一個新的事件流：

// src/js/index.js
// ...
const initialUserInfoSteam = () => {
  const $avator = $(".user_header");
  // 通過頭像 $avator 的 hover 事件來創建流
  const avatorMouseover = Rx.Observable.fromEvent($avator, "mouseover")
    // 500ms 內重復觸發事件則會被忽略
    .debounce(500)
    // 只有當滿足了下列條件的流才會繼續執行，否則將中斷
    .takeWhile((e) => {
      // 異步獲取的用戶信息被新建到 DOM 里，該 DOM 最外層是 infos_container
      // 因此，如果已經有了 infos_container，則可以認為我們已經異步獲取過數據了，此時 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 元素，并插入到頁面中。在這之后，該用戶對應的 DOM 里就會擁有 infos_container 這個 div，所以 takeWhile 會返回 false。也就是說，之后再 hover 上去，流也不會被觸發了
      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

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

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

創建好針對hover的事件流，我們可以把它和上一個事件流結合起來：

// 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);
    });
// ...

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

// 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 里面創建新的流并監聽
      // initialUserInfoSteam($repos);
    })
    // 相反，我們繼續這個流的傳遞，只是通過 flatMap 將原來的流變成了監聽 mouseover 的流
    .flatMap(($repos) => {
      return initialUserInfoSteam($repos);
    });
// ...

DONE ！

栗子中使用到的 RxJS API：

from 通過一個可迭代對象來創建流

fromEvent 通過 DOM 事件來創建流

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

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

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

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

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

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

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

takeWhile 給予流一個判斷，只有當takeWhile中的回調返回true時，流才會繼續執行；否則將中斷之后的事件

What does RxJS observable debounce do

How do I work with jQuery and RxJS

Introduction of observable operators

文章源碼 - rxjs-example