摘要：響應式編程具有很強的表現力，舉個例子來說，限制鼠標重復點擊的例子。在響應式編程中，我把鼠標點擊事件作為一個我們可以查詢和操作的持續的流事件。這在響應式編程中尤其重要，因為我們隨著時間變換會產生很多狀態片段。迭代器模式的另一主要部分來自模式。

Rxjs 響應式編程-第一章：響應式
Rxjs 響應式編程-第二章：序列的深入研究
Rxjs 響應式編程-第三章: 構建并發程序
Rxjs 響應式編程-第四章 構建完整的Web應用程序
Rxjs 響應式編程-第五章 使用Schedulers管理時間
Rxjs 響應式編程-第六章 使用Cycle.js的響應式Web應用程序

現實世界相當混亂：事件不按照順序發生，應用崩潰，網絡不通。幾乎沒有應用是完全同步的，所以我們不得不寫一些異步代碼保持應用的可響應性。大多數的時候是很痛苦的，但也并不是不可避免。

現代應用需要超級快速的響應速度，并且希望能夠不漏掉一個字節的處理來自不同數據源的數據。然而并沒有現成的解決方案，因為它們不會隨著我們添加并發和應用程序狀態而擴展代碼變得越來越復雜。

本章向您介紹反應式編程，這是一種自然，簡單的方法處理異步代碼的方式。我會告訴你事件的流程 - 我們稱之為Observables - 是處理異步代碼的一種很好的方式。
然后我們將創建一個Observable，看看響應式思維和RxJS是怎么樣改善現有技術，讓你成為更快樂，更多高效的程序員。

讓我們從一個小的響應性RxJS程序開始。 這個程序需要通過單擊按鈕檢索來自不同來源的數據，它具有以下要求：

它必須統一來自使用不同源的JSON結構

最終結果不應包含任何副本

為了避免多次請求數據，用戶不能重復點擊按鈕

使用RxJS，我們的代碼類似這樣：

var button = document.getElementById("retrieveDataBtn");
var source1 = Rx.DOM.getJSON("/resource1").pluck("name");
var source2 = Rx.DOM.getJSON("/resource2").pluck("props", "name");
function getResults(amount) {
    return source1.merge(source2)
        .pluck("names")
        .flatMap(function(array) { return Rx.Observable.from(array); })
        .distinct()
        .take(amount);
}
var clicks = Rx.Observable.fromEvent(button, "click");
clicks.debounce(1000)
    .flatMap(getResults(5))
    .subscribe(
        function(value) { console.log("Received value", value); },
        function(err) { console.error(err); },
        function() { console.log("All values retrieved!"); }
    );

不要擔心不理解這里的代碼。只要關注于成果即可。你看到的第一件事是我們使用更少的代碼實現更多的功能。我們通過使用Observable來實現這一目標。

Observable表示數據流。程序也可以可以主要表示為數據流。在前面的示例中，兩個遠程源是Observables，用戶點擊鼠標也是如此。實際上，我們的程序本質上是一個由按鈕的單擊事件構成的Observable，我們把它轉變成獲得我們想要的結果。

響應式編程具有很強的表現力，舉個例子來說，限制鼠標重復點擊的例子。想象一下我們使用我們使用promise和callback實現這個功能是有多復雜：我們需要每秒重置一下點擊次數，并且在用戶點擊之后每秒都要保存點擊狀態。但是這樣子，對于這個小功能來說就顯得過于復雜了，并且所寫代碼與業務功能并沒有直觀的聯系。為了彌補基礎代碼庫的功能不足，在一個大型應用中，這些很小的復雜功能會增加的非常快。

通過響應式編，我們使用debounce方法來限制點擊流次數。這樣就保證每次點擊的間隔時間至少1s，忽略1s之間的點擊次數。我們不關心內部如何實現，我們只是表達我們希望代碼執行的操作，而不是如何操作。

這就變得更有趣了。接下來，您將看到反應式編程如何幫助我們提高課程效率和表現力。

讓我們從這樣一個響應性系統的典型例子開始考慮：點子表格。我們都是使用過吧，但我們很少停下來思考它們是多么令人震驚的直觀。假設我們在電子表格的單元格A1中有一個值，然后我們可以在電子表格中的其他單元格中引用它，并且每當我們更改A1時，每個依賴于A1的單元格都會自動更新與A1同步。

這些操作對我們感覺很自然，我們并不會去告訴計算機去根據A1更新單元格或者如何更新；這些單元格就自動這樣子做了。在點子表格中，我們只需要簡單的聲明我們需要處理的問題，不用操心計算機如何處理。

理解如何把事件作為流，我們回想一下本章開頭的那個程序。在那里，我們使用鼠標點擊作為用戶點擊時實時生成的無限事件流。這個想法起源于Erik Meijer，也就是Rxjs的作者。他認為：你的鼠標就是一個數據庫。

在響應式編程中，我把鼠標點擊事件作為一個我們可以查詢和操作的持續的流事件。想象成流而不是一個孤立的事件，這種想法開辟了一種全新的思考方式。我們可以在其中操縱尚未創建的整個值的流。

好好想想。這種方式有別于我們以往的編程方式，之前我們把數據存儲在數據庫，或者數組并且等待這些數據可用之后在使用它們。如果它們尚不可用(舉個例子：一個網絡請求)，我們只能等它們好了才可以使用。

我們可以將流視為所在由時間而不是存儲位置分開的數組。無論是時間還是存儲位，我們都有元素序列：

將您的程序視為流動的數據序列是理解的RxJS程序的關鍵。這需要一些練習，但并不難。事實上，大多數我們在任何應用程序中使用的數據都可以表示為序列。

讓我們使用傳統javascript傳統的事件綁定技術來實現一個鼠標點擊流。要記錄鼠標點擊的x和y坐標，我們可以這樣寫：

ch1/thinking_sequences1.js

document.body.addEventListener("mousemove", function(e) {
    console.log(e.clientX, e.clientY);
});

此代碼將按順序打印每次鼠標單擊的x坐標和y坐標。

輸出如下：

252 183
211 232
153 323
...

看起來像一個序列，不是嗎？ 當然，問題在于操縱事件并不像操縱數組那么容易。 例如，如果我們想要更改前面的代碼，使其僅記錄前10次位于屏幕右側的單擊事件（相當隨機的目標），我們會寫像這樣的東西：

var clicks = 0;
document.addEventListener("click", function registerClicks(e) {
    if (clicks < 10) {
        if (e.clientX > window.innerWidth / 2) {
            console.log(e.clientX, e.clientY);
            clicks += 1;
        }
    } else {
        document.removeEventListener("click", registerClicks);
    }
});

為了滿足我們的要求，我們通過引入一個全局變量作為擴展狀態來記錄當前點擊數。 我們還需要使用嵌套的條件來檢查兩個不同的條件。當我們完成時，我們必須注銷事件，以免泄漏內存。

副作用和外部狀態
如果一個動作在其發生的范圍之外產生影響，我們稱之為一方副作用。更改函數外部的變量，打印到控制臺或更新數據庫中的值，這些都是副作用。例如改變函數內部的變量是安全的，但是如果該變量超出了我們函數的范圍，那么其他函數也可以改變它的值，這就意味著這個功能不再受控制，因為你無法預測外部會對這個變量作何操作。所以我們需要跟蹤它，添加檢查以確保它的變化符合我們的預期。但是這樣子添加的代碼其實與我們程序無關，確增加程序的復雜度也更容易出錯。雖然副作用總是會有的，但是我們應該努力減少。這在響應式編程中尤其重要，因為我們隨著時間變換會產生很多狀態片段。所以避免外部狀態和副作用是貫穿本書一條宗旨。

我們設法滿足了我們的簡單要求，但是為了實現這樣一個簡單的目標，最終得到了相當復雜的代碼。對于首次查看它的開發人員來說，不容易懂且維護代碼很困難。 更重要的是，因為我們仍然需要保存外部撞他，所以我們很容易在未來發展出玄妙的錯誤。

在這種情況下我們想要的只是查詢點擊的“數據庫”。如果我們是使用關系數據庫，我們使用聲明性語言SQL：

SELECT x, y FROM clicks LIMIT 10

如果我們將點擊事件流視為可以查詢和轉變的數據源，該怎么辦？畢竟，它與數據庫沒有什么不同，都是一個可以處理數據的東西。我們所需要的只是一個為我們提供抽象概念的數據類型。

輸入RxJS及其Observable數據類型：

Rx.Observable.fromEvent(document, "click")
    .filter(function(c) { return c.clientX > window.innerWidth / 2; })
    .take(10)
    .subscribe(function(c) { console.log(c.clientX, c.clientY) })

這段代碼功能同之前，它可以這樣子解讀：

創建一個Observable的點擊事件，并過濾掉在點擊事件上發生屏幕左側的點擊。然后只在控制臺打印前10次點擊的坐標。

注意即使您不熟悉代碼也很容易閱讀，也沒有必要創建外部變量來保持狀態。這樣使我們的代碼是自包含的，不容易產生bug。所以也就沒必要去清除你的狀態。我們可以合并，轉換或者單純的傳遞Observables。我們已經將不容易處理的事件轉變為有形數據結構，這種數據結構與數組一樣易于使用，但更加靈活。

在下一節，我們將看到使Observables如此強大的原理。

要了解Observable的來源，我們需要查看他們的基礎：Observer和Iterator軟件模式。在本節中我們將快速瀏覽它們，然后我們將看到Observables如何結合，簡單而有力。

對于軟件開發人員來說，很難不聽到Observables就想起觀察者模式。在其中我們有一個名為Producer的對象，內部保留訂閱者的列表。當Producer對象發生改變時，訂閱者的update方法會被自動調用。（在觀察者模式的大部分解釋中，這個實體被叫做Subject，為了避免大家和RxJs的自己Subject混淆，我們稱它為Producer）。

ch1/observer_pattern.js

function Producer() {
    this.listeners = [];
}

Producer.prototype.add = function(listener) {
    this.listeners.push(listener);
};

Producer.prototype.remove = function(listener) {
    var index = this.listeners.indexOf(listener);
    this.listeners.splice(index, 1);
};

Producer.prototype.notify = function(message) {
    this.listeners.forEach(function(listener) {
        listener.update(message);
    });
};

Producer對象在實例的偵聽器中保留一個動態的Listener列表，每當Producer更新的時候都會調用其notify方法。在下面的代碼中，我們創建了兩個對象來監聽
notifie，一個Producer的實例。

ch1/observer_pattern.js

// Any object with an "update" method would work.
var listener1 = {
    update: function(message) {
    console.log("Listener 1 received:", message);
}
};
var listener2 = {
    update: function(message) {
    console.log("Listener 2 received:", message);
}
};
var notifier = new Producer();
notifier.add(listener1);
notifier.add(listener2);
notifier.notify("Hello there!");

當我們運行這個程序的時候：

Listener 1 received: Hello there!
Listener 2 received: Hello there!

當notifier更新內部狀態的時候，listener1和listener2都會被更新。這些都不需要我們去操心。

我們的實現很簡單，但它說明了觀察者模式允許觀察者和監聽器解耦。

Observable的另一主要部分來自Iterator模式。一個Iterator是一個為消費者提供簡單的遍象它內容的方式，隱藏了消費者內部的實現。

Iterator接口很簡單。它只需要兩個方法：next（）來獲取序列中的下一個項目，以及hasNext（）來檢查是否還有項目序列。

下面是我們如何編寫一個對數字數組進行操作的迭代器，并且只返回divisor參數的倍數的元素：

ch1/iterator.js

function iterateOnMultiples(arr, divisor) {
    this.cursor = 0;
    this.array = arr;
    this.divisor = divisor || 1;
}
iterateOnMultiples.prototype.next = function() {
    while (this.cursor < this.array.length) {
    var value = this.array[this.cursor++];
    if (value % this.divisor === 0) {
        return value;
    }
}
};
iterateOnMultiples.prototype.hasNext = function() {
    var cur = this.cursor;
    while (cur < this.array.length) {
        if (this.array[cur++] % this.divisor === 0) {
            return true;
        }
    }
    return false;
};

我們可以這樣子使用我們的迭代器：

ch1/iterator.js

var consumer = new iterateOnMultiples([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 3);

console.log(consumer.next(), consumer.hasNext()); // 3 true
console.log(consumer.next(), consumer.hasNext()); // 6 true
console.log(consumer.next(), consumer.hasNext()); // 9 false

迭代器非常適合封裝任何類型數據結構的遍歷邏輯。 正如我們在前面的例子中看到的那樣，迭代器在處理不同類型的數據的時候就會變得很有趣，或者在運行的時候做配置，就像我們在帶有divisor參數的示例中所做的那樣。

雖然Observer和Iterator模式本身就很強大，但是兩者的結合甚至更好。 我們稱之為Rx模式，命名為
在Reactive Extensions庫之后。我們將在本書的其余部分使用這種模式。

Observable序列或簡單的Observable是Rx模式的核心。Observable按順序傳遞出來它的值 - 就像迭代器一樣 - 而不是消費者要求它傳出來的值。這個和觀察者模式有相同之處：得到數據并將它們推送到監聽器。

pull和push
在編程中，基于推送的行為意味著應用程序的服務器組件向其客戶端發送更新，而不是客戶端必須輪詢服務器以獲取這些更新。這就像是說“不要打電話給我們; 我們會打電話給你?！?br>RxJS是基于推送的，因此事件源（Observable）將推動新值給消費者（觀察者），消費者卻不能去主動請求新值。

更簡單地說，Observable是一個隨著時間的推移可以使用其數據的序列。Observables，也就是Observers的消費者相當于觀察者模式中的監聽器。當Observe訂閱一個Observable時，它將在序列中接收到它們可用的值，而不必主動請求它們。

到目前為止，似乎與傳統觀察者沒有太大區別。 但實際上有兩個本質區別：

Observable在至少有一個Observer訂閱它之前不會啟動。

與迭代器一樣，Observable可以在序列完成時發出信號。

使用Observables，我們可以聲明如何對它們發出的元素序列做出反應，而不是對單個項目做出反應。我們可以有效地復制，轉換和查詢序列，這些操作將應用于序列的所有元素。

有幾種方法可以創建Observable，創建函數是最明顯的一種。 Rx.Observable對象中的create方法接受一個Observer參數的回調。 該函數定義了Observable將如何傳出值。這是我們如何創建一個簡單的Observable：

var observable = Rx.Observable.create(function(observer) {
    observer.onNext("Simon");
    observer.onNext("Jen");
    observer.onNext("Sergi");
    observer.onCompleted(); // We are done
});

當我們訂閱此Observable時，它通過在其偵聽器上調用onNext方法來發出三個字符串。 然后它調用onCompleted來表示序列已完成。 但是我們究竟如何訂閱Observable呢？我們使用Observers來做這件事情。

Observers監聽Observables。每當Observable中觸發一個事件，它都會在所有Observers中調用相關的方法。

Observers有三種方法：onNext，onCompleted和onError：

onNext????相當于觀察者模式中的update。 當Observable發出新值時調用它。請注意該名稱如何反映我們訂閱序列的事實，而不僅僅是離散值。

onCompleted????表示沒有更多可用數據。 調用onCompleted后，對onNext的進一步調用將不起作用。

onError????在Observable中發生錯誤時調用。 在調用之后，對onNext的進一步調用將不起作用

以下是我們創建基本觀察者的方法：

var observer = Rx.Observer.create(
    function onNext(x) { console.log("Next: " + x); },
    function onError(err) { console.log("Error: " + err); },
    function onCompleted() { console.log("Completed"); }
);

Rx.Observer對象中的create方法接受onNext，onCompleted和onError情況的函數，并返回一個Observer實例。這三個函數是可選的，您可以決定要包含哪些函數。例如，如果我們訂閱無限序列（例如點擊按鈕（用戶可以永久點擊）），則永遠不會調用onCompleted處理程序。 如果我們確信序列不能出錯（例如，通過從數組中生成一個Observable），我們就不需要onError方法了。

我們還沒有對Observables做過任何實用的事情。如何創建一個檢索遠程內容的Observable？為此，我們將使用Rx.Observable.create包裝XMLHttpRequest對象。

function get(url) {
    return rxjs.Observable.create(function(observer) {
        // Make a traditional Ajax request
        var req = new XMLHttpRequest(); req.open("GET", url);
        req.onload = function() { 
            if (req.status == 200) {
                // If the status is 200, meaning there have been no problems,     
                // Yield the result to listeners and complete the sequence     
                observer.next(req.response);
                observer.completed();
            }
            else {
                // Otherwise, signal to listeners that there has been an error
                observer.error(new Error(req.statusText)); }
            };
            req.onerror = function() {
            observer.error(new Error("Unknown Error"));
        };
        req.send();
    });
}
// Create an Ajax Observable
var test = get("/api/contents.json");

在前面的代碼中，get函數使用create來包裝XMLHttpRequest。如果HTTP GET請求成功，我們emit它的內容并結束序列（我們的Observable只會發出一個結果）。 否則，我們會emit一個錯誤。在最后一行，我們傳入一個url進行調用。 這將創建Observable，但它不會發出任何請求。這很重要：Observable在至少有一個觀察者描述它們之前不會做任何事情。 所以讓我們要這樣子做：

// Subscribe an Observer to it
test.subscribe(
    function onNext(x) { console.log("Result: " + x); }, 
    function onError(err) { console.log("Error: " + err); }, 
    function onCompleted() { console.log("Completed"); }
);

首先要注意的是，我們沒有像之前的代碼那樣顯式創建Observer。大多數時候我們都會使用這個更短的版本，我們在Observable中使用這三個訂閱Observer案例的函數：next，completed和error。

subscribe然后一切就緒。在subscribe之前，我們只是聲明了Observable和Observer將如何交互。只有當我們調用subscribe方法時，一切才開始運行。

在RxJS中，轉換或查詢序列的方法稱為Operator。Operator位于靜態Rx.Observable對象和Observable實例中。在我們的示例中，create就是一個這樣的Operator。

當我們必須創建一個非常具體的Observable時，create是一個很好的選擇，但是RxJS提供了許多其他Operator，可以很容易地為常用源創建Observable。

讓我們再看看前面的例子。對于像Ajax請求這樣的常見操作，通常有一個Operator可供我們使用。 在這種情況下，RxJS DOM庫提供了幾種從DOM相關源創建Observable的方法。由于我們正在執行GET請求，我們可以使用Rx.DOM.Request.get，然后我們的代碼就變成了這個：

Rx.DOM.get("/api/contents.json").subscribe(
    function onNext(data) { console.log(data.response); }, 
    function onError(err) { console.error(err); }
);

rxjs-dom本身支持的rxjs版本比較舊，例子只能做為示意

這段代碼與我們之前的代碼完全相同，但我們不必創建XMLHttpRequest的包裝器： 它已經存在了。另請注意，這次我們省略了onCompleted回調，因為我們不打算在Observable complete時做出反應。我們知道它只會產生一個結果，我們已經在onNext回調中使用它了。

在本書中我們將使用這樣的大量便利操作符。這都是基于rxjs本身的能量，這也正式rxjs強大的地方之一。

一種可以約束全部的數據類型
在RxJS程序中，我們應該努力將所有數據都放在Observables中，而不僅僅是來自異步源的數據。 這樣做可以很容易地組合來自不同來源的數據，例如現有數組與回調結果，或者XMLHttpRequest的結果與用戶觸發的某些事件。
例如，如果我們有一個數組，其項目需要與來自其他地方的數據結合使用，最好將此數組轉換為Observable。（顯然，如果數組只是一個不需要組合的中間變量，則沒有必要這樣做。）在本書中，您將了解在哪些情況下值得將數據類型轉換為Observables。

RxJS為operators提供了從大多數JavaScript數據類型創建Observable的功能。 讓我們回顧一下你將一直使用的最常見的：數組，事件和回調。

我們可以使用通用的operators將任何類似數組或可迭代的對象轉換為Observable。 from將數組作為參數并返回一個包含他所有元素的Observable。

Rx.Observable
.from(["Adrià", "Jen", "Sergi"]) 
.subscribe(
    function(x) { console.log("Next: " + x); }, 
    function(err) { console.log("Error:", err); },
    function() { console.log("Completed"); }
);

from是和fromEvent一起，是RxJS代碼中最方便和最常用的operators之一。

當我們將一個事件轉換為一個Observable時，它就變成了一個可以組合和傳遞的第一類值。 例如，這是一個Observable，只要它移動就會傳初鼠標指針的坐標。

var allMoves = Rx.Observable.fromEvent(document, "mousemove")

allMoves.subscribe(function(e) {
  console.log(e.clientX, e.clientY);
});

將事件轉換為Observable會將事件從之前的事件邏輯中釋放出來。更重要的是，我們可以基于原始的Observables創建新的Observable。這些新的是獨立的，可用于不同的任務。

var movesOnTheRight = allMoves.filter(function(e) { 
    return e.clientX > window.innerWidth / 2;
});
var movesOnTheLeft = allMoves.filter(function(e) { 
    return e.clientX < window.innerWidth / 2;
});
movesOnTheRight.subscribe(function(e) { 
    console.log("Mouse is on the right:", e.clientX);
});
movesOnTheLeft.subscribe(function(e) { 
    console.log("Mouse is on the left:", e.clientX);
});

在前面的代碼中，我們從原始的allMoves中創建了兩個Observable。 這些專門的Observable只包含原始的過濾項：movesOnTheRight包含發生在屏幕右側的鼠標事件，movesOnTheLeft包含發生在左側的鼠標事件。 它們都沒有修改原始的Observable：allMoves將繼續發出所有鼠標移動。 Observable是不可變的，每個應用于它們的operator都會創建一個新的Observable。

如果您使用第三方JavaScript庫，則可能需要與基于回調的代碼進行交互。 我們可以使用fromCallback和fromNodeCallback兩個函數將回調轉換為Observable。Node.js遵循的是在回調函數的第一個參數傳入錯誤對象，表明存在問題。然后我們使用fromNodeCallback專門從Node.js樣式的回調中創建Observable：

var Rx = require("rx"); // Load RxJS
var fs = require("fs"); // Load Node.js Filesystem module
// Create an Observable from the readdir method
var readdir = Rx.Observable.fromNodeCallback(fs.readdir); // Send a delayed message
var source = readdir("/Users/sergi");
var subscription = source.subscribe(
    function(res) { console.log("List of directories: " + res);},
    function(err) { console.log("Error: " + err); },
    function() { console.log("Done!"); 
});

前面的代碼中，我們使用Node.js的fs.readdir方法創建一個Observable readdir。 fs.readdir接受目錄路徑和回調函數delayedMsg，該函數在檢索目錄內容后調用。

我們使用readdir和我們傳遞給原始fs.readdir的相同參數，省掉了回調函數。 這將返回一個Observable，當我們訂閱一個Observer時，它將正確使用onNext，onError和onCompleted。

在本章中，我們探討了響應式編程，并了解了RxJS如何通過Observable解決其他問題的方法，例如callback或promise?，F在您了解為什么Observables功能強大，并且您知道如何創建它們。有了這個基礎，我們現在可以繼續創建更有趣的響應式程序。下一章將向您展示如何創建和組合基于序列的程序，這些程序為Web開發中的一些常見場景提供了更“可觀察”的方法。

關注我的微信公眾號，更多優質文章定時推送

創建了一個程序員交流微信群，大家進群交流IT技術

如果已過期，可以添加博主微信號15706211347，拉你進群