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

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

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

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

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

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

您將要學(xué)習(xí)我們在RxJS程序中最常使用的一些運(yùn)算符。 談?wù)搶π蛄械牟僮骺赡芨杏X很抽象。 為了幫助開發(fā)人員以簡單的方式理解Operator，我們將使用標(biāo)準(zhǔn)的可視化表示序列，稱為大理石圖。 它們直觀地表示異步數(shù)據(jù)流，您可以在RxJS的每個(gè)資源中找到它們。

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

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

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

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

在下面的代碼中，我們將合并兩個(gè)不同的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ùn)算符的大理石圖如下所示：

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

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

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

map是最常用的序列轉(zhuǎn)換運(yùn)算符。它接受一個(gè)Observable和一個(gè)函數(shù)，并將該函數(shù)應(yīng)用于源Observable中的每個(gè)值。 它返回一個(gè)帶有轉(zhuǎn)換值的新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ù)會進(jìn)行一些異步計(jì)算來轉(zhuǎn)換值。在這種情況下，map將無法按預(yù)期工作。 對于這些情況，最好使用flatMap，后續(xù)會介紹到。

filter接受一個(gè)Observable和一個(gè)函數(shù)，并使用該函數(shù)檢測Observable中的每個(gè)元素。它返回一個(gè)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）接受一個(gè)Observable并返回一個(gè)始終包含單個(gè)項(xiàng)的新項(xiàng)，這是在每個(gè)元素上應(yīng)用函數(shù)的結(jié)果。 該函數(shù)接收當(dāng)前元素和函數(shù)先前調(diào)用的結(jié)果。

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是操作序列的強(qiáng)大操作符。事實(shí)上，它是稱為聚合運(yùn)算符的基本實(shí)現(xiàn)。

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

計(jì)算序列的平均值也是一個(gè)聚合操作.RxJS提供了實(shí)例運(yùn)算符的平均值，但是為了本節(jié)的目的，我們想看看如何使用reduce實(shí)現(xiàn)它。每個(gè)聚合運(yùn)算符都可以通過僅使用reduce來實(shí)現(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將每個(gè)新值添加到前一個(gè)值。因?yàn)閞educe不能為我們提供序列中元素的總數(shù)，所以我們需要對它們進(jìn)行計(jì)數(shù)。我們使用包含兩個(gè)字段sum和count的對象組成的初始值調(diào)用reduce，其中我們將存儲到目前為止的元素總數(shù)和總數(shù)。每個(gè)新元素都將返回具有更新值的同一對象。

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

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

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í)間才能完成或無限的序列。在前面的示例中，我們每秒生成一個(gè)增量整數(shù)，并調(diào)用scan替換先前的reduce。我們現(xiàn)在每秒得到生成值的平均值。

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

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

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

flatMap是一個(gè)功能強(qiáng)大的運(yùn)算符，但它比我們迄今為止看到的運(yùn)算符更難理解?？梢园阉胂蟪蒓bservables的concatAll()。

concatAll是一個(gè)函數(shù)，它接受一個(gè)數(shù)組數(shù)組并返回一個(gè)“flattened”單個(gè)數(shù)組，其中包含所有子數(shù)組的值，而不是子數(shù)組本身。 我們可以使用reduce來實(shí)現(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ù)組變扁平。它需要一個(gè)源Observable和一個(gè)返回一個(gè)新的Observable的函數(shù)，并將該函數(shù)應(yīng)用于源Observable中的每個(gè)元素，就像map一樣。如果程序在這里停止，我們最終會得到一個(gè)會發(fā)出Observables的Observable。 但是flatMap向主序列發(fā)出每個(gè)新Observable發(fā)出的值，將所有Observable“扁平化”為一個(gè)主序列。 最后，我們獲得了一個(gè)Observable。

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

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

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

在下面的示例中，我們將兩個(gè)Observers訂閱到計(jì)數(shù)器Observable，它每秒發(fā)出一個(gè)遞增的整數(shù)。 兩秒后，我們?nèi)∠诙€(gè)訂閱，我們可以看到它的輸出停止但第一個(gè)訂閱者的輸出繼續(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ù)時(shí)候，Operater會自動(dòng)取消訂閱。當(dāng)序列結(jié)束或滿足操作條件時(shí)，range或take等操作符將取消訂閱。更高級的操作符，如withLatestFrom或flatMapLatest，將根據(jù)需要在內(nèi)部創(chuàng)建和銷毀訂閱，因?yàn)樗鼈兲幚淼氖沁\(yùn)行中的幾個(gè)可觀察的內(nèi)容。簡而言之，大部分訂閱的取消都不應(yīng)該是你該擔(dān)心的。

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

在下面的代碼中，我們嘗試取消對包含promise p的Observable的訂閱，同時(shí)我們以傳統(tǒng)的方式設(shè)置一個(gè)動(dòng)作來解決promise。 promise應(yīng)在五秒內(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方法仍在運(yùn)行，表明取消Observable并不會取消關(guān)聯(lián)的Promsie。

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

我們不能在回調(diào)中使用傳統(tǒng)的try / catch機(jī)制，因?yàn)樗峭降摹?它將在任何異步代碼之前運(yùn)行，并且無法捕獲任何錯(cuò)誤。

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

讓我們看看如何捕獲Observables中的錯(cuò)誤。

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

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

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

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

我們將帶有三個(gè)JSON字符串的數(shù)組傳遞給getJSON，其中數(shù)組中的第二個(gè)字符串包含語法錯(cuò)誤，因此JSON.parse將無法解析它。 然后我們將訂閱結(jié)果，為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為第一個(gè)結(jié)果發(fā)出解析的JSON，但在嘗試解析第二個(gè)結(jié)果時(shí)拋出異常。 onError處理程序捕獲并打印出來。默認(rèn)行為是，每當(dāng)發(fā)生錯(cuò)誤時(shí)，Observable都會停止發(fā)出項(xiàng)目，并且不會調(diào)用onCompleted。

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

catch接受一個(gè)Observable或一個(gè)接收錯(cuò)誤的函數(shù)作為參數(shù)并返回另一個(gè)Observable。 在我們的場景中，如果原始Observable中存在錯(cuò)誤，我們希望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)建了一個(gè)新的Observable，它使用catch運(yùn)算符來捕獲原始Observable中的錯(cuò)誤。 如果出現(xiàn)錯(cuò)誤，它將使用僅發(fā)出一個(gè)項(xiàng)目的Observable繼續(xù)序列，并使用描述錯(cuò)誤的error屬性。 這是輸出：

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

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

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

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

有時(shí)錯(cuò)誤就會發(fā)生，我們無能為力。例如，可能存在請求遠(yuǎn)程數(shù)據(jù)的超時(shí)，因?yàn)橛脩艟哂胁环€(wěn)定的Internet連接，或者我們查詢的遠(yuǎn)程服務(wù)器可能崩潰。在這些情況下，如果我們能夠繼續(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)建了一個(gè)函數(shù)，該函數(shù)返回一個(gè)Observable，它使用XMLHttpRequest從URL檢索內(nèi)容。 因?yàn)槲覀兊倪B接可能有點(diǎn)不穩(wěn)定，所以我們在訂閱它之前添加retry（5），確保在出現(xiàn)錯(cuò)誤的情況下，它會在放棄并顯示錯(cuò)誤之前嘗試最多五次。

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

其次，重試將始終重新嘗試整個(gè)Observable序列，即使某些項(xiàng)目沒有錯(cuò)誤。如果您在處理項(xiàng)目時(shí)造成任何副作用，這一點(diǎn)很重要，因?yàn)槊看沃卦嚩紩匦聭?yīng)用它們。

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

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

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

examples_earthquake/index.html

現(xiàn)在我們的HTML已準(zhǔn)備就緒，我們可以為我們的應(yīng)用程序編寫邏輯。首先，我們需要知道我們獲得了什么樣的數(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ù)組包含一個(gè)對象，其中包含今天發(fā)生的每次地震的數(shù)據(jù)。 那是一大堆數(shù)據(jù)！ 一天之內(nèi)發(fā)生了多少次地震是令人驚訝的（并且可怕）。對于我們的程序，我們只需要每次地震的坐標(biāo)，標(biāo)題和大小。

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

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);
});

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

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

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

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

這是可以使用flatMap的完美情況。我們將使用Rx.Observable.from檢索數(shù)據(jù)并從features數(shù)組中生成一個(gè)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);
});

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

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

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

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

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

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

到目前為止，我們已經(jīng)使用了rx.all.js中包含的RxJS運(yùn)算符，但通常還是需要借鑒其他基于RxJS的庫附帶的運(yùn)算符。在我們的例子中，我們將看看RxJS-DOM。RxJS-DOM是一個(gè)外部庫，其中包含一個(gè)處理JSONP請求的運(yùn)算符：jsonpRequest。這為我們節(jié)省了一些代碼，因?yàn)槲覀儾恍枰褂糜憛挼娜趾瘮?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);
});

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

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

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)建一個(gè)Observable，每隔五秒發(fā)出一個(gè)遞增的數(shù)字。我們對這些數(shù)字沒有做任何事情; 相反，我們使用flatMap來檢索jsonpRequest的數(shù)據(jù)。另請注意我們?nèi)绾卧谑紫葯z索列表時(shí)出現(xiàn)問題時(shí)再次嘗試重試。

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

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

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

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

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

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

Rx.Observable.from

默認(rèn)行為：同步

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

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

Rx.Observable.range

默認(rèn)行為：同步

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

Rx.Observable.interval

默認(rèn)行為：異步

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

Rx.Observable.distinct

默認(rèn)行為：與filter的Observable相同

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

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

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

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

關(guān)注我的微信公眾號，更多優(yōu)質(zhì)文章定時(shí)推送