本文是一篇 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'));  

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

1. 4.x 穩定版 Document
2. 5 beta 版 Document
3. 從 4 到 5 的遷移

Let's start

如上所說，我們要利用 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訪問到。

初始化 DOM 事件流

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

<!-- 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 事件來創建流 
  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，那么就可以通過它來異步獲取數據。將整個過程拆分一下：

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

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

1. 讓每個 value 返回一個 Observable
2. 通過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變量來判斷是否正在異步，如果正在異步，則不進行新的異步;而我們更希望的是能夠取消舊的異步，只處理新的異步請求。

更加優雅的 Rx 風格

按照上面的教程，我們在 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事件創建一個流。用戶的詳細資料是異步加載的，而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 !

APIS

栗子中使用到的 RxJS API：

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