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

本例的所有代碼在 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'));  

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

1. 4.x 穩(wěn)定版 Document
2. 5 beta 版 Document
3. 從 4 到 5 的遷移

Let's start

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

通過RxJS，在輸入過程中實時進(jìn)行異步搜索： 

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訪問到。

初始化 DOM 事件流

在index.html中編寫一個input，我們將在index.js中，通過 RxJS 的 Observable 監(jiān)聽input的keyup事件?？梢允褂胒romEvent來創(chuàng)建一個基于 DOM 事件的流，并通過map和filter進(jìn)一步處理。

<!-- index.html --> 
<input class="search" type="text" maxlength="1000" required placeholder="search in github"/> 
// 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)前的值。

實時進(jìn)行異步獲取

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

1. 用戶在 input 里輸入任意內(nèi)容
2. 觸發(fā)keyup事件，獲取到當(dāng)前 value
3. 將 value 代入到一個異步方法里，通過接口獲取數(shù)據(jù)
4. 利用返回數(shù)據(jù)渲染 DOM

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

1. 讓每個 value 返回一個 Observable
2. 通過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ā)出多個異步事件之后，每個事件所耗費(fèi)的時間不一定相同。如果前一個異步所用時間較后一個長，那么當(dāng)它最終返回結(jié)果時，有可能把后面的異步率先返回的結(jié)果覆蓋

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

優(yōu)化事件流

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

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

也就是說，當(dāng)用戶在連續(xù)打字時，我們不應(yīng)該繼續(xù)進(jìn)行之后的事件處理，而如果打字中斷，或者說兩次keyup事件的時間間隔足夠長時，才應(yīng)該發(fā)送異步請求。針對這點(diǎn)，可以使用 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ù)事件中的第一個)就好了?？梢岳胐istinctUntilChanged方法： 

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

發(fā)出多個異步事件之后，每個事件所耗費(fèi)的時間不一定相同。如果前一個異步所用時間較后一個長，那么當(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); 
// ...  

流的監(jiān)聽

至此，我們對 input keyup以及異步獲取數(shù)據(jù)的整個事件流處理完畢，并進(jìn)行了一定的優(yōu)化，避免了過多的請求、異步返回結(jié)果錯亂等問題。但創(chuàng)建了一個流之后也有對其進(jìn)行監(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變量來判斷是否正在異步，如果正在異步，則不進(jìn)行新的異步;而我們更希望的是能夠取消舊的異步，只處理新的異步請求。

更加優(yōu)雅的 Rx 風(fēng)格

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

但是有一個問題：小應(yīng)用的另一個功能是，當(dāng)鼠標(biāo)hover到頭像上時，異步獲取并展現(xiàn)用戶的信息?？墒怯脩纛^像是在subscribe回調(diào)中動態(tài)插入的，又該如何創(chuàng)建事件流呢?當(dāng)然了，可以在每次插入 DOM 之后在利用fromEvent創(chuàng)建一個基于hover的事件流，但那樣總是不太好的，寫出來的代碼也不夠 Rx?；蛟S我們就不應(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 中實際上什么都不用做，就能達(dá)到之前的效果 
observable.subscribe(() => { 
  console.log('success'); 
}, (err) => { 
  console.log(err); 
}, () => { 
  console.log('completed'); 
});  

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

創(chuàng)建基于hover的事件流

我們接下來針對用戶頭像的hover事件創(chuàng)建一個流。用戶的詳細(xì)資料是異步加載的，而hover到頭像上時彈出 modal。如果是第一個hover，則 modal 里只有一個 loading 的圖標(biāo)，并且異步獲取數(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，則可以認(rèn)為我們已經(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時，流可以正常進(jìn)行;而一旦返回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 !

APIS

栗子中使用到的 RxJS API：

• from 通過一個可迭代對象來創(chuàng)建流
• fromEvent 通過 DOM 事件來創(chuàng)建流
• debounce 如果在一定時間內(nèi)流中的某個事件不斷被觸發(fā)，則不會進(jìn)行之后的事件操作
• map 遍歷流中所有事件，返回新的流
• filter 篩選流中所有事件，返回新的流
• flatMap 對各個事件返回的值進(jìn)行處理并返回 Observable，然后將所有的 Observable 扁平化，成為一個新的 Observable
• flatMapLatest 對各個事件返回的值進(jìn)行處理并返回 Observable，然后將所有的 Observable 扁平化，成為一個新的 Observable。但只會獲取最后一次返回的 Observable，其他的返回結(jié)果不予處理
• distinctUntilChanged 流中如果相鄰事件的結(jié)果一樣，則僅篩選出一個(剔除重復(fù)值)
• do 可以依次拿到流上每個事件的返回值，利用其做一些無關(guān)流傳遞的事情
• takeWhile 給予流一個判斷，只有當(dāng)takeWhile中的回調(diào)返回true時，流才會繼續(xù)執(zhí)行;否則將中斷之后的事件