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

本例的所有代碼在 github 倉(cāng)庫(kù)：rxjs-example

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

ReactiveX - rxjs RxJS 5 beta 版

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

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

# 安裝 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í)可以代入一個(gè)對(duì)象作為參數(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 來一步步做一個(gè) github 小應(yīng)用。首先完成其基本功能，即通過一個(gè) input 輸入文字，并實(shí)時(shí)根據(jù) input 內(nèi)值的變化去發(fā)送異步請(qǐng)求，調(diào)用 github API 進(jìn)行搜索。如圖所示(線上 Demo)：

通過RxJS，在輸入過程中實(shí)時(shí)進(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，我們將會(huì)啟動(dòng)一個(gè) 8080 端口的服務(wù)器，使得我們編譯好的資源可以在localhost:8080/webpack-dev-server訪問到。

初始化 DOM 事件流

在index.html中編寫一個(gè)input，我們將在index.js中，通過 RxJS 的 Observable 監(jiān)聽input的keyup事件。可以使用fromEvent來創(chuàng)建一個(gè)基于 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 時(shí)搜索框的值，篩選出合法值 
      .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時(shí)在 console 里輸出 input 當(dāng)前的值。

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

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

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

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

1. 讓每個(gè) value 返回一個(gè) Observable
2. 通過flatMap將所有的 Observable 扁平化，成為一個(gè)新的 Observable

圖解flatMap： 

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

// src/js/helper.js 
const SEARCH_REPOS = 'https://api.github.com/search/repositories?sort=stars&order=desc&q='; 
 
// 創(chuàng)建一個(gè) ajax 的 promise 
const getReposPromise = (query) => { 
  return $.ajax({ 
      type: "GET", 
    url: `${SEARCH_REPOS}${query}`, 
  }).promise(); 
}; 
// 通過 fromPromise 創(chuàng)建一個(gè) 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 方法將返回一個(gè) Observable 
    // flatMap 則將所有 Observable 合并，轉(zhuǎn)為一個(gè) Observable 
    .flatMap(getRepos); 
// ...  

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

• 不斷打字時(shí)會(huì)連續(xù)不斷觸發(fā)異步請(qǐng)求，占用資源影響體驗(yàn)
• 如果相鄰的keyup事件觸發(fā)時(shí) input 的值一樣，也就是說按下了不改變 value 的按鍵(比如方向鍵)，會(huì)重復(fù)觸發(fā)一樣的異步事件
• 發(fā)出多個(gè)異步事件之后，每個(gè)事件所耗費(fèi)的時(shí)間不一定相同。如果前一個(gè)異步所用時(shí)間較后一個(gè)長(zhǎng)，那么當(dāng)它最終返回結(jié)果時(shí)，有可能把后面的異步率先返回的結(jié)果覆蓋

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

優(yōu)化事件流

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

不斷打字時(shí)會(huì)連續(xù)不斷觸發(fā)異步請(qǐng)求，占用資源影響體驗(yàn)

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

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

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

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

也就是說，對(duì)于任意相鄰的事件，如果它們的返回值一樣，則只要取一個(gè)(重復(fù)事件中的第一個(gè))就好了。可以利用distinctUntilChanged方法： 

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

這個(gè)蛋疼的問題我相信大家很可能遇見過。在發(fā)送多個(gè)異步請(qǐng)求時(shí)，因?yàn)樗脮r(shí)長(zhǎng)不一定，無法保障異步返回的先后順序，所以，有時(shí)候可能早請(qǐng)求的異步的結(jié)果會(huì)覆蓋后來請(qǐng)求的異步結(jié)果。

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

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

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

至此，我們對(duì) input keyup以及異步獲取數(shù)據(jù)的整個(gè)事件流處理完畢，并進(jìn)行了一定的優(yōu)化，避免了過多的請(qǐng)求、異步返回結(jié)果錯(cuò)亂等問題。但創(chuàng)建了一個(gè)流之后也有對(duì)其進(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); 
// 第一個(gè)回調(diào)中的 data 代表異步的返回值 
observable.subscribe((data) => { 
  // 在 showNewResults 方法中使用返回值渲染 DOM 
  showNewResults(data); 
}, (err) => { 
  console.log(err); 
}, () => { 
  console.log('completed'); 
}); 
 
// 異步返回的結(jié)果是個(gè) Array，代表搜索到的各個(gè)倉(cāng)庫(kù) item 
// 遍歷所有 item，轉(zhuǎn)化為 jQuery 對(duì)象，最后插入到 content_container 中 
const showNewResults = (items) => { 
  const repos = items.map((item, i) => { 
    return reposTemplate(item); 
  }).join(''); 
  $('.content_container').html(repos); 
};  

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

而如果我們不使用 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)行新的異步;而我們更希望的是能夠取消舊的異步，只處理新的異步請(qǐng)求。

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

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

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

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

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

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 合并到原有的流中。此時(shí)流中的每個(gè)元素是 results 中的每個(gè) item 
    .flatMap((results) => Rx.Observable.from(results)) 
    // 將各 item 轉(zhuǎn)化為 jQuery 對(duì)象 
    .map((repos) => $(reposTemplate(repos))) 
    // 最后把每個(gè) jQuery 對(duì)象依次加到容器里 
    .do(($repos) => { 
      $conatiner.append($repos); 
    }); 
 
// 在 subscribe 中實(shí)際上什么都不用做，就能達(dá)到之前的效果 
observable.subscribe(() => { 
  console.log('success'); 
}, (err) => { 
  console.log(err); 
}, () => { 
  console.log('completed'); 
});  

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

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

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

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

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

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

// 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ā)事件則會(huì)被忽略 
    .debounce(500) 
    // 只有當(dāng)滿足了下列條件的流才會(huì)繼續(xù)執(zhí)行，否則將中斷 
    .takeWhile((e) => { 
      // 異步獲取的用戶信息被新建到 DOM 里，該 DOM 最外層是 infos_container 
      // 因此，如果已經(jīng)有了 infos_container，則可以認(rèn)為我們已經(jīng)異步獲取過數(shù)據(jù)了，此時(shí) takeWhile 將返回 false，流將會(huì)中斷 
      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è)面中。在這之后，該用戶對(duì)應(yīng)的 DOM 里就會(huì)擁有 infos_container 這個(gè) div，所以 takeWhile 會(huì)返回 false。也就是說，之后再 hover 上去，流也不會(huì)被觸發(fā)了 
      const {data, conatiner} = result; 
      showUserInfo(conatiner, data); 
    }); 
 
  avatorMouseover.subscribe((result) => { 
      console.log('fetch user info succeed'); 
  }, (err) => { 
    console.log(err); 
  }, () => { 
    console.log('completed'); 
  }); 
};  

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

由圖可知，當(dāng)takeWhile中的回調(diào)返回true時(shí)，流可以正常進(jìn)行;而一旦返回false，則之后的事件不會(huì)再發(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)建好針對(duì)hover的事件流，我們可以把它和上一個(gè)事件流結(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)可以使用了，但依舊不夠好。目前總共有兩個(gè)流：監(jiān)聽 input keyup的流和監(jiān)聽mouseover的流。但是，因?yàn)橛脩纛^像是動(dòng)態(tài)插入的 ，所以我們必須在$conatiner.append($repos);之后才能創(chuàng)建并監(jiān)聽mouseover。不過鑒于我們已經(jīng)在最后的do方法里插入了獲取的數(shù)據(jù)，所以可以試著把兩個(gè)流合并到一起：

// src/js/index.js 
// ... 
const initialUserInfoSteam = ($repos) => { 
  const $avator = $repos.find('.user_header'); 
  const avatorMouseover = Rx.Observable.fromEvent($avator, 'mouseover') 
  // ... 流的處理跟之前的一樣 
  // 但我們不再需要 subscribe 它，而是返回這個(gè) Observable 
  return avatorMouseover; 
}; 
 
const observable = Rx.Observable.fromEvent($input, 'keyup') 
    // ... 
    .do(($repos) => { 
      $conatiner.append($repos); 
      // 不再在 do 里面創(chuàng)建新的流并監(jiān)聽 
      // initialUserInfoSteam($repos); 
    }) 
    // 相反，我們繼續(xù)這個(gè)流的傳遞，只是通過 flatMap 將原來的流變成了監(jiān)聽 mouseover 的流 
    .flatMap(($repos) => { 
      return initialUserInfoSteam($repos); 
    }); 
// ...  

DONE !

APIS

栗子中使用到的 RxJS API：

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