摘要：但不同的是，在的遍歷調(diào)用過程中，如果一個事件還沒有觸發(fā)完畢獲取到返回值，就觸發(fā)了下一個事件，則將忽略返回的值。這樣，我們就可以避免異步的返回值因為返回較慢，反而覆蓋了之后異步的返回值。

ReactiveX，又稱 Reactive Extensions（響應(yīng)式擴(kuò)展），其中的 X 代表各種語言。因為它實(shí)質(zhì)上只是一個事件流處理庫，不需要什么其他依賴。

講一講 ReactiveX 中的 “流”：

我的理解是，Rx 通過 “流” 的概念，將事件串聯(lián)成一個個事件流，各個事件流之間還可進(jìn)行 "并聯(lián)" 的作用。當(dāng)某個流上的事件被調(diào)用時，就可以觸發(fā)我們設(shè)定好的監(jiān)聽回調(diào)。

那么什么樣的事件可以成為流呢？答案是任何事件。無論是異步非阻塞事件（setTimeOut、網(wǎng)絡(luò)請求等），還是同步可阻塞事件（點(diǎn)擊事件、對迭代器的遍歷等），一切都是流。此時不得不祭出一張神棍圖：

事件的串聯(lián)是流。比方說，用戶對一個按鈕進(jìn)行了猛烈的點(diǎn)擊，所有的點(diǎn)擊事件就是一個流；再或者，并發(fā)多個網(wǎng)絡(luò)請求，它們也是一個流。而 Rx 的主要作用，就是為流的處理提供了一整套的解決方案：將不同的流進(jìn)行組合，或者監(jiān)聽事件的觸發(fā)及時給予響應(yīng)等等。

RxJS 的作者曾在 The introduction to Reactive Programming you"ve been missing 一文中詳細(xì)講解了 RxJS 的初步使用和流的概念，我們先僅看文中的一幅圖來理解 Rx 的概念：

這是一個多次點(diǎn)擊事件形成的事件流，在從左往右的時間線上有很多個點(diǎn)擊事件，而每個點(diǎn)擊事件的時間間隔各不相同。通過 Rx 我們可以對流上的各個事件進(jìn)行篩選，并獲取到在某一段時間內(nèi)的連續(xù)點(diǎn)擊次數(shù)。

作者自己吐槽過那篇文章實(shí)在太長，所以又有了后來的這篇 2 minute introduction to Rx 文章。在文章中有如下解釋：

我們在頁面上的點(diǎn)擊事件就可以組成流。比如一個記錄每次點(diǎn)擊時坐標(biāo)的流。隨機(jī)點(diǎn)擊頁面多次之后，可能會產(chǎn)生這樣一個流：

而我們可以通過 RxJS 中的方法對這個流內(nèi)的各個事件進(jìn)行篩選，比如選出橫坐標(biāo) x < 250 的點(diǎn)擊：

filter( (event) -> event.x < 250 )

也就是說，我們可以像對待 JavaScript 中可遍歷對象一樣，對流上的各個事件進(jìn)行遍歷，選出符合條件的事件。這就是 Rx 的魅力所在。

既然 Rx 是為了流而生的，那么最佳運(yùn)用場景當(dāng)然是面對一系列較復(fù)雜的事件流時了。

比如，用戶在一個 input 內(nèi)輸入文字。每次keyup的時候就會根據(jù)輸入內(nèi)容，請求 Wikipedia 的 API 進(jìn)行搜索：

var input = document.getElementById("input");
// 通過 fromEvent 以及 input keyup 事件創(chuàng)建一個流
var dictionarySuggest = Rx.Observable.fromEvent(input, "keyup")
  // 獲取到每次 keyup 時的input value，并通過 filter 保證其合法性
  .map(function () { return input.value; })
  .filter(function (text) { return !!text; })
  .distinctUntilChanged()
  .debounce(250)
  // searchWikipedia 為異步請求方法
  .flatMapLatest(searchWikipedia)
  .subscribe(
      // onNext
    function (results) {
      list = [];
      list.concat(results.map(createItem));
    },
    // onError
    function (err) {
      logError(err);
    }
  );

我們創(chuàng)建了一個流來處理從用戶keyup，到searchWikipedia，再到處理網(wǎng)絡(luò)請求結(jié)果這一系列事件，并且在其中對事件進(jìn)行了篩選判斷：

filter 剔除掉不合法的值

distinctUntilChanged 當(dāng)用戶按下例如 左、右 這種按鈕時，不會改變 input 的值，但也會觸發(fā)keyup事件。這種時候就完全沒有必要重復(fù)發(fā)送異步請求了。distinctUntilChanged會剔除流中有著相同的值的元素

debounce 在過了一段指定的時間還沒觸發(fā)事件時才觸發(fā)下一個事件。也就是說，在打字過程中，如果用戶在指定事件間隔（250ms）內(nèi)沒有再打字，則觸發(fā)下一個事件（searchWikipedia）；否則我們認(rèn)為用戶在連續(xù)打字，所以不會頻繁的發(fā)送網(wǎng)絡(luò)請求

flatMapLatest

首先，它是一個flatMap方法。它用一個指定的函數(shù)（searchWikipedia）對原始流中的每一項數(shù)據(jù)執(zhí)行變換操作，并返回一個Observable，flatMap將所有的返回值組成一個新的流。

其次，flatMapLatest擁有flatMap的全部特性。但不同的是，在flatMapLatest的遍歷調(diào)用過程中，如果一個事件 A 還沒有觸發(fā)完畢獲取到返回值，就觸發(fā)了下一個事件 B，則將忽略 A 返回的值。這樣，我們就可以避免 A 異步的返回值因為返回較慢，反而覆蓋了之后 B 異步的返回值。用圖解釋如下：

subscribe 創(chuàng)建對流的監(jiān)聽，并提供了成功和失敗的回調(diào)

而在傳統(tǒng)的編寫方法里，我們可能會創(chuàng)建 input 的keyup監(jiān)聽事件，并緩存上一次的值；每次keyup時，要判斷當(dāng)前值是否合法，并且與上一次的值不一樣。除此以外，還要創(chuàng)建一個定時器，每隔一段時間就用合法的值去請求searchWikipedia方法 --- 即便這樣，也無法保證不在用戶連續(xù)打字時發(fā)送請求。

可以看到，在我們把事件串成流并進(jìn)行處理之后，要比傳統(tǒng)的編寫方式方便很多。

假設(shè)我們要讀取一個 4GB 的大文件，將其加密后寫入到一個新文件里。直接將整個文件讀到內(nèi)存里再加密、寫入肯定是不行的，反之，我們依賴 RxJS 的流，創(chuàng)建多個讀取、加密、寫入事件，形成三個流出來：

文件讀取流：每次調(diào)用方法時異步讀取 64k 的文件

加密流：對讀取的文件進(jìn)行加密

寫入流：將加密好的內(nèi)容異步寫入新文件

最后對整個observable進(jìn)行監(jiān)聽

var fs = require("fs");
var Rx = require("rx");

// Read/write from stream implementation
function readAsync(fd, chunkSize) { /* impl */ }
function appendAsync(fd, buffer) { /* impl */ }
function encrypt(buffer) { /* impl */}

// 打開一個 4GB 的文件，每次只讀取 64k
var inFile = fs.openSync("4GBfile.txt", "r+");
var outFile = fs.openSync("Encrypted.txt", "w+");

readAsync(inFile, 2 << 15)
  .map(encrypt)
  .flatMap(function (data) {
    return appendAsync(outFile, data);
  })
  .subscribe(
      // onNext
    function () { },
    // onError
    function (err) {
      console.log("An error occurred while encrypting the file: %s", err.message);
      fs.closeSync(inFile);
      fs.closeSync(outFile);
    },
    // onCompleted
    function () {
      console.log("Successfully encrypted the file.");
      fs.closeSync(inFile);
      fs.closeSync(outFile);
    }
  );

由此可以看出，在應(yīng)對較復(fù)雜的事件流或者處理多個異步事件的時候，使用 RxJS 會有一定優(yōu)勢；但如果復(fù)雜度沒有這么高的時候則沒有太大的使用必要。

目前為止，本文基本介紹了 RxJS 的核心概念 --- 針對事件流的管理與掌控。

在下一篇文章里，我們將會利用 RxJS，完成一個簡單的 github 應(yīng)用。戳：探索 RxJS - 做一個 github 小應(yīng)用可查看文章，rxjs-example查看案例源碼。