摘要：接下來，我們將實現(xiàn)一個真實的應用程序，顯示幾乎實時發(fā)生的地震。得到的由表示，其中包含和的合并元素。如果不同同時傳出元素，合并序列中這些元素的順序是隨機的。是操作序列的強大操作符。但是的方法仍在運行，表明取消并不會取消關聯(lián)的。

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

童年的回憶中的益智視頻游戲，你必須使用各種技巧在屏幕上引導下降的水流。您可以拆分流，稍后將它們合并，或者使用傾斜的木板來改變它們的方向。你必須要有創(chuàng)造力才能使水達到最終目標。

我發(fā)現(xiàn)該游戲與使用Observable序列有很多相似之處。 Observable只是我們可以轉換，組合和查詢的事件流。 無論我們是在處理簡單的Ajax回調還是在Node.js中處理字節(jié)數(shù)據(jù)都沒關系。 我們發(fā)現(xiàn)流的方式是一樣的。 一旦我們在流中思考，我們程序的復雜性就會降低。

在本章中，我們將重點介紹如何在程序中有效地使用序列。 到目前為止，我們已經(jīng)介紹了如何創(chuàng)建Observable并使用它們進行簡單的操作。為了釋放它們的力量，我們必須知道將我們的程序輸入和輸出轉換為帶有我們程序流程的序列。

在我們弄清楚之前，我們將會遇到一些可以幫助我們開始操作序列的基本operator。接下來，我們將實現(xiàn)一個真實的應用程序，顯示（幾乎）實時發(fā)生的地震。 開始吧！

您將要學習我們在RxJS程序中最常使用的一些運算符。 談論對序列的操作可能感覺很抽象。 為了幫助開發(fā)人員以簡單的方式理解Operator，我們將使用標準的可視化表示序列，稱為大理石圖。 它們直觀地表示異步數(shù)據(jù)流，您可以在RxJS的每個資源中找到它們。

讓我們使用范圍運算符，它返回一個Observable，它得到指定范圍內(nèi)的整數(shù)：Rx.Observable.range（1,3）;

它的大理石圖看起來像這樣:

長箭頭表示Observable，x軸表示時間。每個圓圈表示Observable通過內(nèi)部調用onNext（）傳出的值。生成第三個值后，range調用了onCompleted，在圖中用垂直線表示。

讓我們看一個涉及幾個Observable的例子。合并運算符采用兩個不同的Observable并返回一個具有合并值的新Observable。 interval運算符返回一個Observable，它在給定的時間間隔內(nèi)產(chǎn)生增量數(shù)，以毫秒為單位。

在下面的代碼中，我們將合并兩個不同的Observable，它們使用interval來以不同的間隔生成值：

var a = Rx.Observable.interval(200).map(function(i) { 
    return "A" + i;
});
var b = Rx.Observable.interval(100).map(function(i) {
    return "B" + i; 
    
});
Rx.Observable.merge(a, b).subscribe(function(x) {
    console.log(x);
});

B0, A0, B1, B2, A1, B3, B4...

合并運算符的大理石圖如下所示：

這里，沿y軸的虛線箭頭指向應用于序列A和B中每個元素的變換的最終結果。得到的Observable由C表示，其中包含A和B的合并元素。如果不同Observables同時傳出元素，合并序列中這些元素的順序是隨機的。

在RxJS中轉換Observables的數(shù)十個運算符中，最常用的是具有良好收集處理能力的其他語言也具有：map，filter和reduce。在JavaScript中，您可以在Array中找到這些operator。

RxJS遵循JavaScript約定，因此您會發(fā)現(xiàn)以下運算符的語法與數(shù)組運算符的語法幾乎相同。實際上，我們將使用數(shù)組和Observables同時實現(xiàn)，以顯示兩個API的相似程度。

map是最常用的序列轉換運算符。它接受一個Observable和一個函數(shù)，并將該函數(shù)應用于源Observable中的每個值。 它返回一個帶有轉換值的新Observable。

JS Arrays

var src = [1, 2, 3, 4, 5];
var upper = src.map(function(name) {
    return name * 2; 
});
upper.forEach(logValue);

Observables

var src = Rx.Observable.range(1, 5); 
var upper = src.map(function(name) {
    return name * 2; 
});
upper.subscribe(logValue);

在這兩種情況下，src都不會發(fā)生改變。

這段代碼和后面的代碼使用的logValue函數(shù)：

var logValue = function(val) { 
    console.log(val) 
};

有些情況下，我們傳遞給map的函數(shù)會進行一些異步計算來轉換值。在這種情況下，map將無法按預期工作。 對于這些情況，最好使用flatMap，后續(xù)會介紹到。

filter接受一個Observable和一個函數(shù)，并使用該函數(shù)檢測Observable中的每個元素。它返回一個Observable序列，其中包含函數(shù)返回true的所有元素。

JS Arrays

var isEven = (function(val) { return val % 2 !== 0; });
var src = [1, 2, 3, 4, 5];
var even = src.filter(isEven);
even.forEach(logValue);

Observables

var isEven = (function(val) { return val % 2 !== 0; });
var src = Rx.Observable.range(1, 5); 
var even = src.filter(isEven);
even.subscribe(logValue);

reduce（也稱為fold）接受一個Observable并返回一個始終包含單個項的新項，這是在每個元素上應用函數(shù)的結果。 該函數(shù)接收當前元素和函數(shù)先前調用的結果。

JS Arrays

var src = [1, 2, 3, 4, 5];
var sum = src.reduce(function(a, b) {
    return a + b;
});
console.log(sum);

Observables

var src = Rx.Observable.range(1, 5);
var sum = src.reduce(function(acc, x) {
    return acc + x;
});
sum.subscribe(logValue);

reduce是操作序列的強大操作符。事實上，它是稱為聚合運算符的基本實現(xiàn)。

聚合運算符處理序列并返回單個值。例如， Rx.Observable.first接受一個Observable和一個可選函數(shù)，并返回滿足函數(shù)條件布爾值的第一個元素。

計算序列的平均值也是一個聚合操作.RxJS提供了實例運算符的平均值，但是為了本節(jié)的目的，我們想看看如何使用reduce實現(xiàn)它。每個聚合運算符都可以通過僅使用reduce來實現(xiàn)：

sequences/marble.js

var avg = Rx.Observable.range(0, 5)
    .reduce(function(prev, cur) {
        return {
            sum: prev.sum + cur,
            count: prev.count + 1
        };
    }, { sum: 0, count: 0 })
    .map(function(o) {
        return o.sum / o.count;
    });
    
var subscription = avg.subscribe(function(x) {
    console.log("Average is: ", x);
});

Average is: 2

在此代碼中，我們使用reduce將每個新值添加到前一個值。因為reduce不能為我們提供序列中元素的總數(shù)，所以我們需要對它們進行計數(shù)。我們使用包含兩個字段sum和count的對象組成的初始值調用reduce，其中我們將存儲到目前為止的元素總數(shù)和總數(shù)。每個新元素都將返回具有更新值的同一對象。

當序列結束時，reduce可以通過調用onNex返回t包含最終總和和最終計數(shù)的對象。但在這里我們使用map來返回將總和除以計數(shù)的結果。

我們可以聚合無限Observables嗎？
想象一下，我們正在編寫一個程序，讓用戶在行走時獲得平均速度。即使用戶尚未完成行走，我們也需要能夠使用我們目前所知的速度值進行計算。我們想要實時記錄無限序列的平均值。 問題是如果序列永遠不會結束，像reduce這樣的聚合運算符將永遠不會調用其Observers的onNext運算符。
對我們來說幸運的是，RxJS團隊已經(jīng)考慮過這種情況，并為我們提供了scan操作符，其作用類似于reduce但是會發(fā)出每個中間結果：

var avg = Rx.Observable.interval(1000)
    .scan(function (prev, cur) {
        return {
            sum: prev.sum + cur,
            count: prev.count + 1
        };
    }, { sum: 0, count: 0 })
    .map(function(o) {
        return o.sum / o.count;
    });
    
var subscription = avg.subscribe( function (x) {
    console.log(x);
});
這樣，我們可以聚合需要很長時間才能完成或無限的序列。在前面的示例中，我們每秒生成一個增量整數(shù)，并調用scan替換先前的reduce。我們現(xiàn)在每秒得到生成值的平均值。

如果你的Observable的結果是還是Observables，你要怎么處理？大多數(shù)情況下，您希望在單個序列中統(tǒng)一這些嵌套Observable中的項目。 這正是flatMap的作用。

flatMap運算符接收參數(shù)Observable A，其元素也是Observables，并返回一個子元素也是Observable的Observable。讓我們用圖表可視化它：

我們可以看到A（A1，A2，A3）中的每個元素也是可觀察序列。 一旦我們使用變換函數(shù)將flatMap應用于A，我們得到一個Observable，其中包含A的不同子元素中的所有元素。

flatMap是一個功能強大的運算符，但它比我們迄今為止看到的運算符更難理解。可以把它想象成Observables的concatAll()。

concatAll是一個函數(shù)，它接受一個數(shù)組數(shù)組并返回一個“flattened”單個數(shù)組，其中包含所有子數(shù)組的值，而不是子數(shù)組本身。 我們可以使用reduce來實現(xiàn)這樣的功能：

function concatAll(source) {
    return source.reduce(function(a, b) {
        return a.concat(b); 
    });
}

我們會像這樣使用它：

concatAll([[0, 1, 2], [3, 4, 5], [6, 7, 8]]);
// [0, 1, 2, 3, 4, 5, 6, 7, 8]

flatMap做同樣的事情，但它使Observables而不是數(shù)組變扁平。它需要一個源Observable和一個返回一個新的Observable的函數(shù)，并將該函數(shù)應用于源Observable中的每個元素，就像map一樣。如果程序在這里停止，我們最終會得到一個會發(fā)出Observables的Observable。 但是flatMap向主序列發(fā)出每個新Observable發(fā)出的值，將所有Observable“扁平化”為一個主序列。 最后，我們獲得了一個Observable。

在RxJS中，我們可以取消正在運行的Observable。 這是一種優(yōu)于其他異步通信形式的優(yōu)勢，例如回調和Promise，一旦被調用就無法直接取消（盡管某些Promise實現(xiàn)支持取消）。

我們可以通過兩種主要方式取消Observable：隱式和顯式。

Observables本身沒有取消的方法。相反，當我們訂閱Observable時，我們會得到一個代表該特定訂閱的Disposable對象。然后我們可以在該對象中調用方法dispose，并且該訂閱將停止從Observable接收通知。

在下面的示例中，我們將兩個Observers訂閱到計數(shù)器Observable，它每秒發(fā)出一個遞增的整數(shù)。 兩秒后，我們?nèi)∠诙€訂閱，我們可以看到它的輸出停止但第一個訂閱者的輸出繼續(xù)：

sequences/disposable.js

var counter = Rx.Observable.interval(1000);

var subscription1 = counter.subscribe(function(i) {
    console.log("Subscription 1:", i);
});

var subscription2 = counter.subscribe(function(i) {
    console.log("Subscription 2:", i);
});

setTimeout(function() { 
    console.log("Canceling subscription2!");
    subscription2.dispose();
}, 2000);

Subscription 1: 0 
Subscription 2: 0 
Subscription 1: 1 
Subscription 2: 1 
Canceling subscription2! 
Subscription 1: 2 
Subscription 1: 3 
Subscription 1: 4
...

大多數(shù)時候，Operater會自動取消訂閱。當序列結束或滿足操作條件時，range或take等操作符將取消訂閱。更高級的操作符，如withLatestFrom或flatMapLatest，將根據(jù)需要在內(nèi)部創(chuàng)建和銷毀訂閱，因為它們處理的是運行中的幾個可觀察的內(nèi)容。簡而言之，大部分訂閱的取消都不應該是你該擔心的。

當您使用包含不提供取消的外部API的Observable時，Observable仍會在取消時停止發(fā)出通知，但基礎API不一定會被取消。例如，如果您正在使用封裝Promise的Observable，則Observable將在取消時停止發(fā)出，但不會取消基礎Promise。

在下面的代碼中，我們嘗試取消對包含promise p的Observable的訂閱，同時我們以傳統(tǒng)的方式設置一個動作來解決promise。 promise應在五秒內(nèi)resolve，但我們在創(chuàng)建后立即取消訂閱：

var p = new Promise(function(resolve, reject) {
    window.setTimeout(resolve, 5000);
});

p.then(function() {
    console.log("Potential side effect!");
});

var subscription = Rx.Observable.fromPromise(p).subscribe(function(msg) {
    console.log("Observable resolved!");
});

subscription.dispose();

5秒后，我們看到：

Potential side effect!

如果我們?nèi)∠麑bservable的訂閱，它會有效地阻止它接收通知。 但是promise的then方法仍在運行，表明取消Observable并不會取消關聯(lián)的Promsie。

了解我們在Observable中使用的外部API的詳細信息非常重要。您可能認為已取消序列，但底層API會繼續(xù)運行并在程序中引起一些副作用。 這些錯誤真的很難捕捉到。

我們不能在回調中使用傳統(tǒng)的try / catch機制，因為它是同步的。 它將在任何異步代碼之前運行，并且無法捕獲任何錯誤。

在回調函數(shù)中，可以通過將錯誤（如果有）作為參數(shù)傳遞到回調函數(shù)。這是有用的，但它使代碼非常脆弱。

讓我們看看如何捕獲Observables中的錯誤。

還記得我們在上面上討論了第一次與觀察者聯(lián)系的觀察者可以調用的三種方法嗎？ 我們熟悉onNext和onCompleted，但是我們還沒有使用onError; 它是有效處理Observable序列中錯誤的關鍵。

為了了解它是如何工作的，我們將編寫一個簡單的函數(shù)來獲取JSON字符串數(shù)組，并使用JSON.parse返回一個Observable，它發(fā)出從這些字符串解析的對象：

為了了解它是如何工作的，我們將編寫一個簡單的函數(shù)來獲取JSON字符串組成的數(shù)組，并使用JSON.parse返回一個Observable，它發(fā)出從這些字符串解析的對象：

function getJSON(arr) {
    return Rx.Observable.from(arr).map(function(str) {
        var parsedJSON = JSON.parse(str);
        return parsedJSON;
    });
}

我們將帶有三個JSON字符串的數(shù)組傳遞給getJSON，其中數(shù)組中的第二個字符串包含語法錯誤，因此JSON.parse將無法解析它。 然后我們將訂閱結果，為onNext和onError提供處理程序:

getJSON([
    "{"1": 1, "2": 2}",
    "{"success: true}", // Invalid JSON string
    "{"enabled": true}"
]).subscribe(
    function(json) {
        console.log("Parsed JSON: ", json);
    },
    function(err) {
        console.log(err.message);
    }
)

Parsed JSON: { 1: 1, 2: 2 }
JSON.parse: unterminated string at line 1 column 8 of the JSON data

Observable為第一個結果發(fā)出解析的JSON，但在嘗試解析第二個結果時拋出異常。 onError處理程序捕獲并打印出來。默認行為是，每當發(fā)生錯誤時，Observable都會停止發(fā)出項目，并且不會調用onCompleted。

到目前為止，我們已經(jīng)看到如何檢測錯誤已經(jīng)發(fā)生并對該信息做了些什么，但是我們無法對它做出響應并繼續(xù)我們正在做的事情。Observable察實例具有catch運算符，它允許我們對Observable中的錯誤做出反應并繼續(xù)使用另一個Observable。

catch接受一個Observable或一個接收錯誤的函數(shù)作為參數(shù)并返回另一個Observable。 在我們的場景中，如果原始Observable中存在錯誤，我們希望Observable發(fā)出包含error屬性的JSON對象：

function getJSON(arr) {
    return Rx.Observable.from(arr).map(function(str) {
        var parsedJSON = JSON.parse(str);
        return parsedJSON;
    });
}

var caught = getJSON(["{"1": 1, "2": 2}", "{"1: 1}"]).catch(
    Rx.Observable.return({
        error: "There was an error parsing JSON"
    })
);

caught.subscribe(
    function(json) {
        console.log("Parsed JSON: ", json);
    },
    // Because we catch errors now, `onError` will not be executed
    function(e) {
        console.log("ERROR", e.message);
    }
);

在前面的代碼中，我們創(chuàng)建了一個新的Observable，它使用catch運算符來捕獲原始Observable中的錯誤。 如果出現(xiàn)錯誤，它將使用僅發(fā)出一個項目的Observable繼續(xù)序列，并使用描述錯誤的error屬性。 這是輸出：

Parsed JSON: Object { 1: 1, 2: 2 }
Parsed JSON: Object { error: "There was an error parsing JSON" }

這是catch操作符的大理石圖：

注意X表示序列出錯。 在這種情況下，Observable值 - 三角形的不同形狀意味著它們是來自另一個Observable的值。在這里，這是我們在發(fā)生錯誤時返回的Observable。

catch對于對序列中的錯誤作出反應非常有用，它的行為與傳統(tǒng)的try / catch塊非常相似。 但是，在某些情況下，忽略Observable中的項目發(fā)生的錯誤并讓序列繼續(xù)，這將是非常方便的。 在這些情況下，我們可以使用重試運算符。

有時錯誤就會發(fā)生，我們無能為力。例如，可能存在請求遠程數(shù)據(jù)的超時，因為用戶具有不穩(wěn)定的Internet連接，或者我們查詢的遠程服務器可能崩潰。在這些情況下，如果我們能夠繼續(xù)請求我們需要的數(shù)據(jù)直到成功，那將是很好的。 重試操作符的確如此：

sequences/error_handling.js

// This will try to retrieve the remote URL up to 5 times.
Rx.DOM.get("/products").retry(5)
.subscribe(
    function(xhr) { console.log(xhr); },
    function(err) { console.error("ERROR: ", err); }
);

在前面的代碼中，我們創(chuàng)建了一個函數(shù)，該函數(shù)返回一個Observable，它使用XMLHttpRequest從URL檢索內(nèi)容。 因為我們的連接可能有點不穩(wěn)定，所以我們在訂閱它之前添加retry（5），確保在出現(xiàn)錯誤的情況下，它會在放棄并顯示錯誤之前嘗試最多五次。

使用重試時需要了解兩件重要事項。首先，如果我們不傳遞任何參數(shù)，它將無限期地重試，直到序列完成沒有錯誤。 如果Observable產(chǎn)生錯誤，這對性能是危險的。 如果我們使用同步Observable，它將具有與無限循環(huán)相同的效果。

其次，重試將始終重新嘗試整個Observable序列，即使某些項目沒有錯誤。如果您在處理項目時造成任何副作用，這一點很重要，因為每次重試都會重新應用它們。

使用我們在本章中到目前為止所涵蓋的概念，我們將構建一個使用RxJS的Web應用程序，以向我們展示實時發(fā)生地震的位置。我們首先要建立一個功能性的反應性實施方案，我們將隨著時間的推移對其進行改進。 最終結果如下：

我們將使用USGS（美國地質調查局）地震數(shù)據(jù)庫，該數(shù)據(jù)庫提供多種格式的實時地震數(shù)據(jù)集。 我們將以JSONP格式從每周數(shù)據(jù)集中獲取數(shù)據(jù)。

我們還將使用Leaflet（一個JavaScript庫）來渲染交互式地。讓我們看看我們的index.html看起來如何，并重點介紹：

examples_earthquake/index.html

現(xiàn)在我們的HTML已準備就緒，我們可以為我們的應用程序編寫邏輯。首先，我們需要知道我們獲得了什么樣的數(shù)據(jù)以及在地圖上代表地震所需什么樣的數(shù)據(jù)。

USGS網(wǎng)站給我們的JSONP數(shù)據(jù)看起來像這樣：

examples_earthquake/jsonp_example.txt

eqfeed_callback({
    "type": "FeatureCollection",
    "metadata": {
        "generated": 1408030886000,
        "url": "http://earthquake.usgs.gov/earthquakes/...",
        "title": "USGS All Earthquakes, Past Day",
        "status": 200, "api": "1.0.13", "count": 134
    },
    "features": [
        {
            "type": "Feature",
            "properties": {
                "mag": 0.82,
                "title": "M 0.8 - 3km WSW of Idyllwild-Pine Cove, California",
                "place": "3km WSW of Idyllwild-Pine Cove, California",
                "time": 1408030368460,
                ...
            },
            "geometry": {
                "type": "Point",
                "coordinates": [ -116.7636667, 33.7303333, 17.33 ]
            },
            "id": "ci15538377"
        },
        ...
    ]
})

features數(shù)組包含一個對象，其中包含今天發(fā)生的每次地震的數(shù)據(jù)。 那是一大堆數(shù)據(jù)！ 一天之內(nèi)發(fā)生了多少次地震是令人驚訝的（并且可怕）。對于我們的程序，我們只需要每次地震的坐標，標題和大小。

我們首先要創(chuàng)建一個Observable來檢索數(shù)據(jù)集并發(fā)出單個地震。 這是第一個版本：

examples_earthquake/code.js

var quakes = Rx.Observable.create(function(observer) {
    window.eqfeed_callback = function(response) {
        var quakes = response.features;
        quakes.forEach(function(quake) {
            observer.onNext(quake);
        });
    };
    loadJSONP(QUAKE_URL);
});

quakes.subscribe(function(quake) {
    var coords = quake.geometry.coordinates;
    var size = quake.properties.mag * 10000;
    L.circle([coords[1], coords[0]], size).addTo(map);
});

等等，那個明顯的全局函數(shù)window.eqfeed_callback在我們的代碼中做了什么？ 好吧，事實證明，JSONP URL通常在URL中添加查詢字符串，以指定處理響應的函數(shù)名稱，但USGS站點不允許這樣做，因此我們需要創(chuàng)建一個全局函數(shù) 他們決定我們必須使用的名稱，即eqfeed_callback。

我們的Observable按順序發(fā)出所有地震。我們現(xiàn)在有地震數(shù)據(jù)生成器！我們不必關心異步流程或者必須將所有邏輯放在同一個函數(shù)中。只要我們訂閱Observable，就會得到地震數(shù)據(jù)。

通過在地震觀測中將地震檢索“黑箱”，我們現(xiàn)在可以訂閱并處理每次地震。 然后我們將為每個地震繪制一個圓，其大小與其大小成比例。

我們可以做得更好嗎？你打賭！在前面的代碼中，我們?nèi)匀煌ㄟ^遍歷數(shù)組并調用onNext來管理每個地震，即使我們在Observable中將其隔離。

這是可以使用flatMap的完美情況。我們將使用Rx.Observable.from檢索數(shù)據(jù)并從features數(shù)組中生成一個Observable。 然后我們將Observable合并回主Observable中：

var quakes = Rx.Observable.create(function(observer) {
    window.eqfeed_callback = function(response) {
        observer.onNext(response);
        observer.onCompleted();
    };
    loadJSONP(QUAKE_URL);
}).flatMap(function transform(dataset) {
    return Rx.Observable.from(dataset.response.features);
});

quakes.subscribe(function(quake) {
    var coords = quake.geometry.coordinates;
    var size = quake.properties.mag * 10000;
    L.circle([coords[1], coords[0]], size).addTo(map);
});

我們不再手動管理流程了。 沒有循環(huán)或條件來提取單個地震對象并將其傳遞出去。 這是就是發(fā)生了什么：

onNext只發(fā)生一次，它產(chǎn)生整個JSON字符串。

由于我們只會產(chǎn)生一次，因此我們在onNext之后發(fā)出完成信號。

我們將flatMap調用鏈接到create的結果，因此flatMap將從Observable中獲取每個結果（在這種情況下只有一個），將它用作transform函數(shù)的參數(shù)，并將該函數(shù)產(chǎn)生的Observable合并到源Observable。

這里我們采用包含所有地震的features數(shù)組，并從中創(chuàng)建一個Observable。由于flatMap，這將成為quakes變量將包含的實際Observable。

5.訂閱不會改變; 它像以前一樣繼續(xù)處理地震的數(shù)據(jù)流。

到目前為止，我們已經(jīng)使用了rx.all.js中包含的RxJS運算符，但通常還是需要借鑒其他基于RxJS的庫附帶的運算符。在我們的例子中，我們將看看RxJS-DOM。RxJS-DOM是一個外部庫，其中包含一個處理JSONP請求的運算符：jsonpRequest。這為我們節(jié)省了一些代碼，因為我們不需要使用討厭的全局函數(shù)：

examples_earthquake/code1_2.js

var quakes = Rx.DOM.jsonpRequest({
    url: QUAKE_URL,
    jsonpCallback: "eqfeed_callback"
})
.flatMap(function(result) {
    return Rx.Observable.from(result.response.features);
})
.map(function(quake) {
    return {
        lat: quake.geometry.coordinates[1],
        lng: quake.geometry.coordinates[0],
        size: quake.properties.mag * 10000
    };
});

quakes.subscribe(function(quake) {
    L.circle([quake.lat, quake.lng], quake.size).addTo(map);
});

請記住，要運行此代碼，您需要在HTML中包含RxJS-DOM中的文件rx.dom.js。請注意我們?nèi)绾翁砑右粋€map運算符，將地震對象轉換為僅包含我們可視化所需信息的簡單對象：緯度，經(jīng)度和地震震級。 我們在subscribeoperator中寫的功能越少越好。

我們地震應用的版本不會實時更新地震圖。為了實現(xiàn)這一點，我們將使用我們在本章前面看到的interval運算符 - 以及有用的distinct運算符。下面的代碼，然后我們將完成更改：

examples_earthquake/code1_3.js

var quakes = Rx.Observable
.interval(5000)
.flatMap(function() {
    return Rx.DOM.jsonpRequest({
        url: QUAKE_URL,
        jsonpCallback: "eqfeed_callback"
    }).retry(3);
})
.flatMap(function(result) {
    return Rx.Observable.from(result.response.features);
})
.distinct(function(quake) { return quake.properties.code; });

quakes.subscribe(function(quake) {
    var coords = quake.geometry.coordinates;
    var size = quake.properties.mag * 10000;
    L.circle([coords[1], coords[0]], size).addTo(map);
});

在前面的代碼中，我們使用interval來發(fā)出新請求并以5秒的固定間隔處理它們。 interval創(chuàng)建一個Observable，每隔五秒發(fā)出一個遞增的數(shù)字。我們對這些數(shù)字沒有做任何事情; 相反，我們使用flatMap來檢索jsonpRequest的數(shù)據(jù)。另請注意我們?nèi)绾卧谑紫葯z索列表時出現(xiàn)問題時再次嘗試重試。

我們應用的最后一個運算符是distinct，它只發(fā)出之前未發(fā)出的元素。 它需要一個函數(shù)來返回屬性以檢查是否相等。 這樣我們就不會重繪已經(jīng)繪制過的地震。

在不到20行中，我們編寫了一個應用程序，定期輪詢外部JSONP URL，從其內(nèi)容中提取具體數(shù)據(jù)，然后過濾掉已導入的地震。在那之后，我們在地圖上表示地震，其大小與其大小成比例-所有這些都以獨立，清晰和簡潔的方式編寫，而不依賴于外部狀態(tài)。這表明了Observables的表現(xiàn)力。

這里有一些想法可以使用你新獲得的RxJS技能，并使這個小應用程序更有趣：

當用戶將鼠標懸停在地震上時，提供一個彈出窗口，顯示有關該特定地震的更多信息。 一種方法是從只有你想要顯示的屬性的地震中創(chuàng)建一個新的Observable，并在懸停時動態(tài)過濾它。

在頁面頂部放置一個計數(shù)器，顯示當前到目前為止的地震次數(shù)，并每天重置

本章向您介紹了一些新的運算符，所以這里是對它們的回顧，以及我們在應用程序中使用它們的方法。 請記住，您始終可以在RxJS GitHub站點上找到Operator的完整API文檔。

Rx.Observable.from

默認行為：同步

由于您在應用程序中使用的許多數(shù)據(jù)源都來自數(shù)組或迭代器，因此有一個運算符可以從中創(chuàng)建Observable。 from是您最常使用的Operator之一。

使用from，我們可以從數(shù)組，類似數(shù)組的對象（例如，arguments對象或DOM NodeLists）創(chuàng)建Observable，甚至可以實現(xiàn)可迭代協(xié)議的類型，例如String，Map和Set

Rx.Observable.range

默認行為：同步

range運算符生成有限的Observable，它發(fā)出特定范圍內(nèi)的整數(shù)。它功能多樣，可用于許多場景。 例如，您可以使用范圍在像掃雷一樣的游戲板上生成初始方塊。

Rx.Observable.interval

默認行為：異步

每次需要生成時間間隔的值時，您可能會以interval運算符作為生成器開始。由于interval每x毫秒發(fā)出一次順序整數(shù)（其中x是我們傳遞的參數(shù)），我們只需要將值轉換為我們想要的任何值。 我們在第3章“構建并發(fā)程序”中的游戲很大程度上基于該技術。

Rx.Observable.distinct

默認行為：與filter的Observable相同

distinct是這些非常簡單的Operator之一，可以節(jié)省大量的開發(fā)工作。它會過濾掉已經(jīng)發(fā)出的任何值。 這使我們避免編寫容易出錯的樣板代碼，我們將對比傳入的結果決定返回值。就是返回不同值。

distinct允許我們使用指定比較方法的函數(shù)。另外，我們可以不傳遞任何參數(shù)，它將使用嚴格的比較來比較數(shù)字或字符串等基本類型，并在更復雜的對象的情況下運行深度比較。

在本章中，我們介紹了如何使用大理石圖表直觀地表示和理解Observable流程。我們已經(jīng)介紹了最常見的運算符來轉換Observables，更重要的是，我們只使用Observable序列構建了一個真實的世界應用程序，避免設置任何外部狀態(tài)，循環(huán)或條件分支。我們以聲明的方式表達了我們的整個程序，而不必編碼完成手頭任務的每一步。

在下一章中，我們將繼續(xù)探索Observable序列，這次我們將介紹更高級的運算符，它們允許您控制程序中的流和數(shù)據(jù)，用之前無法想象的代碼！

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