RxJS 教程

KnewOne 發(fā)布于2019-08-19 17:22 / 2927人閱讀

摘要：是針對異步數(shù)據(jù)流的編程。所以這個數(shù)據(jù)流只包含一個簡單的反射值。且慢，天生就是處理異步數(shù)據(jù)流的，為何不把請求的響應(yīng)作為一個攜帶數(shù)據(jù)的流呢么么噠，概念上沒有問題，我們就來操作一下。你需要明確通知觀察者或者訂閱者數(shù)據(jù)流的到達或者錯誤的發(fā)生。

"Reactive Programming是神馬？"

互聯(lián)網(wǎng)上有很多不是很友好的解釋。維基百科寬泛而玄乎。 Stackoverflow教科書式的解釋非常不適合信任Reactive Manifesto 聽起來像是給給項目經(jīng)理或者是銷售的匯報。微軟的 Rx 定義 "Rx = Observables + LINQ + Schedulers" 太重并且太微軟化了，讓人看起來不知所云?！绊憫?yīng)”、“變化發(fā)生”這些術(shù)語無法很好地闡釋Reactive Programming的顯著特點，聽起來和你熟悉的MV*、編程語言差別不大。當(dāng)然，我的視角也是基于模型和變換的，要是脫離了這些概念，一切都是無稽之談了。

那么我要開始吧啦吧啦了，(后文中，將使用RP代替Reactive Programming，私底下譯者將Reactive Programming，翻譯為響應(yīng)式編程)。

RP 是針對異步數(shù)據(jù)流的編程。

一定程度而言，RP并不算新的概念。Event Bus、點擊事件都是異步流。開發(fā)者可以觀測這些異步流，并調(diào)用特定的邏輯對它們進行處理。使用Reactive如同開掛：你可以創(chuàng)建點擊、懸停之類的任意流。通常流廉價(點擊一下就出來一個)而無處不在，種類豐富多樣：變量，用戶輸入，屬性，緩存，數(shù)據(jù)結(jié)構(gòu)等等都可以產(chǎn)生流。舉例來說：微博回文(譯者注：比如你關(guān)注的微博更新了)和點擊事件都是流：你可以監(jiān)聽流并調(diào)用特定的邏輯對它們進行處理。

基于流的概念，RP賦予了你一系列神奇的函數(shù)工具集，使用他們可以合并、創(chuàng)建、過濾這些流。 一個流或者一系列流可以作為另一個流的輸入。你可以合并兩個流，從一堆流中過濾你真正感興趣的那一些，將值從一個流映射到另一個流。

如果流是RP的核心，我們不妨從“點擊頁面中的按鈕”這個熟悉的場景詳細地了解它。

流是包含了有時序，正在進行事件的序列，可以發(fā)射(emmit)值(某種類型)、錯誤、完成信號。流在包含按鈕的瀏覽器窗口被關(guān)閉時發(fā)出完成信號。

我們異步地捕獲發(fā)射的事件，定義一系列函數(shù)在值被發(fā)射后，在錯誤被發(fā)射后，在完成信號被發(fā)射后執(zhí)行。有時，我們忽略對錯誤，完成信號地處理，僅僅關(guān)注對值的處理。對流進行監(jiān)聽，通常稱為訂閱，處理流的函數(shù)是觀測者，流是被觀測的主體。這就是觀測者設(shè)計模式。

教程中，我們有時會使用ASCII字符來繪制圖表：

--a---b-c---d---X---|->

a, b, c, d 是數(shù)據(jù)流發(fā)射的值
X 是數(shù)據(jù)流發(fā)射的錯誤
| 是完成信號
---> 是時序軸

嗶嗶完了，我們來點新的，不然很快你就感覺到寂寞了。我們將把原來的點擊事件流轉(zhuǎn)換為新的點擊事件流。

首先我們創(chuàng)建一個計數(shù)流來表明按鈕被點擊的次數(shù)。在RP中，每一個流都擁有一系列方法，例如map，filter，scan 等等。當(dāng)你在流上調(diào)用這些方法，例如clickStream.map(f)，會返回基于點擊事件流的新的流，同時原來的點擊事件流并不會被改變，這個特性被稱為不可變性(immutability)。不可變性與RP配合相得益彰，如同美酒加咖啡。我們可以鏈?zhǔn)降卣{(diào)用他們：clickStream.map(f).scan(g)

  clickStream: ---c----c--c----c------c--->
               vvvvv map(c becomes 1) vvvvv
               ---1----1--1----1------1--->
               vvvvvvvvv  scan(+) vvvvvvvvv
counterStream: ---1----2--3----4------5--->

map(f) 函數(shù)對原來的流使用我們出入的f函數(shù)進行轉(zhuǎn)換，并生成新的流。在上面的例子中，我們將每一次點擊映射為數(shù)字1。scan(g)函數(shù)將所有流產(chǎn)生的值進行匯總，通過傳入x = g(accumulated, current)函數(shù)產(chǎn)生新的值，g 是簡單的求和函數(shù)。最后 counterStream在點擊發(fā)生后發(fā)射點擊事件發(fā)生的總數(shù)。

為了展示Reactive的真正力量，我們舉個例子：你想要“兩次點擊”事件的流，或者是“三次點擊”，或者是n次點擊的流。深呼吸一下，試著想想怎么用傳統(tǒng)的命令、狀態(tài)式方法來解決。我打賭這個這會相當(dāng)操蛋，你會搞些變量來記錄狀態(tài)，還要搞些處理時延的機制。

如果用RP來解決，太他媽簡單了。實際上4行代碼就可以搞定。先不要看代碼，不管你是菜鳥還是牛逼，使用圖表來思考可以使你更好地理解構(gòu)建這些流的方法。

灰色框里面的函數(shù)會把一個流轉(zhuǎn)換成另外一個流。首先我們把點擊打包到list中，如果點擊后消停了250毫秒，我們就重新打包一個新的list(顯然buffer(stream.throttle(250ms))就是用來干這個的，不明白細節(jié)沒有關(guān)系，反正是demo嘛)。我們在列表上調(diào)用map()，將列表的長度映射為一個整數(shù)的流。最后，我們通過filter(x >= 2)過濾掉整數(shù)1。哈哈：3個操作就生成了我們需要的流，現(xiàn)在我們可以訂閱(監(jiān)聽)這個流，然后來完成我們需要的邏輯了。

通過這個例子，我希望你能感受到使用RP的牛逼之處了。這僅僅是冰山一角。你可以在不同地流上(比如API響應(yīng)的流)進行同樣的操作。同時，Reactive還提供了許多其他實用的函數(shù)。

"我要在今后采用RP范式進行編程嗎？"

RP 提高了編碼的抽象程度，你可以更好地關(guān)注在商業(yè)邏輯中各種事件的聯(lián)系避免大量細節(jié)而瑣碎的實現(xiàn)，使得編碼更加簡潔。

使用RP，將使得數(shù)據(jù)、交互錯綜復(fù)雜的web、移動app開發(fā)收益更多。10年以前，與網(wǎng)頁的交互僅僅是提交表單、然后根據(jù)服務(wù)器簡單地渲染返回結(jié)果這些事情。App進化得越來越有實時性：修改表單中一個域可以同步地更新到后端服務(wù)器?！包c贊”信息實時地在不同用戶設(shè)備上同步。

現(xiàn)代App中大量的實時事件創(chuàng)造了更好的交互和用戶體驗，披荊斬棘需要利劍在手，RP就是你手中的利劍。

通過實例RP編程思想

我們將從實例可以深入RP的編程思想，文章末尾，一個完整地實例應(yīng)用會被構(gòu)建，你也會理解整個過程。

我選擇 JavaScript 和 RxJS 作為實例的構(gòu)建工具。因為大多開發(fā)者都熟悉JavaScript語言。Rx* library family 在各種語言和平臺都是實現(xiàn) (.NET, Java, Scala, Clojure, JavaScript, Ruby, Python, C++, Objective-C/Cocoa, Groovy, 等等)。無論你選擇在哪個平臺或者那種語言實踐RP，你都將從本教程中受益。(譯者注：Rx，即ReactiveX，其中X代表不同的語言和技術(shù)棧，比如.NET，Java，Scala，Ruby，Javascript。RxJS表示RP基于Javascript語言的實現(xiàn)。后文中Rx代表所有實現(xiàn)了RP的特定技術(shù)棧)

微博(Twitter)簡易版“你可能感興趣的人”

微博主頁，有一個組件會推薦給你那些你可能感興趣的人。

我們的Demo將使用這個場景，關(guān)注下面這些主要特性：

頁面打開后，通過API加載數(shù)據(jù)展示3個你可能感興趣的用戶賬號

點擊“刷新”按鈕，重新加載三個新的用戶賬號

在一個用戶賬號上點擊"x" 按鈕，清除當(dāng)前這個賬戶，重新加載一個新的賬戶

每行展示賬戶的信息和這個賬戶主頁的鏈接

其他特性和按鈕我們暫且忽略，由于Twitter在最近關(guān)閉了公共API授權(quán)接口，我們選擇Github作為代替，展示GitHub用戶的賬戶。實例中我們使用該接口獲取GitHub用戶.

如果你希望先睹為快，完成后的代碼已經(jīng)發(fā)布在了Jsfiddle。

"你可能感興趣的用戶"請求&響應(yīng)

這個問題使用Rx怎么解?，呵呵，我們從Rx的箴言開始： 神馬都是流 。首先我們做最簡單的部分——頁面打開后通過API加載3個賬戶的信息。分三步走：(1)發(fā)一個請求(2)獲得響應(yīng)(3)依據(jù)響應(yīng)渲染頁面。那么，我們先使用流來表示請求。我靠，表示個請求用得著嗎？不過千里之行始于足下。

頁面加載時，僅需要一個請求。所以這個數(shù)據(jù)流只包含一個簡單的反射值。稍后，我們再研究如何多個請求出現(xiàn)的情況，現(xiàn)在先從一個請求開始。

--a------|->

a是字符串 "https://api.github.com/users"

這個流中包含了我們希望請求的URL地址。一旦這個請求事件發(fā)生，我們可以獲知兩件事情：請求流發(fā)射值(字符串URL)的時間就是請求需要被執(zhí)行的時間，請求需要請求的地址就是請求流發(fā)射的值。

在Rx*中構(gòu)建一個單值的流很容易。官方術(shù)語中把流稱為“觀察的對象”("Observable")，因為流可以被觀察、訂閱，這么稱呼顯得很蠢，我自己把他們稱為 stream 。

var requestStream = Rx.Observable.just("https://api.github.com/users");

目前這個攜帶字符串的流沒有其他操作,我們需要在這個流發(fā)射值之后，做點什么：通過訂閱這個流來實現(xiàn)。

requestStream.subscribe(function(requestUrl) {
  // 執(zhí)行異步請求
  jQuery.getJSON(requestUrl, function(responseData) {
    // ...
  });
}

我們采用了jQuery的Ajax回調(diào) (假設(shè)讀著已經(jīng)了解jQuery ajax回調(diào)) 來處理異步請求操作。且慢，Rx天生就是處理異步數(shù)據(jù)流的，
為何不把請求的響應(yīng)作為一個攜帶數(shù)據(jù)的流呢？么么噠，概念上沒有問題，我們就來操作一下。

requestStream.subscribe(function(requestUrl) {
  // 執(zhí)行異步請求
  var responseStream = Rx.Observable.create(function (observer) {
    jQuery.getJSON(requestUrl)
    .done(function(response) { observer.onNext(response); })
    .fail(function(jqXHR, status, error) { observer.onError(error); })
    .always(function() { observer.onCompleted(); });
  });
  
  responseStream.subscribe(function(response) {
    // 業(yè)務(wù)邏輯
  });
}

使用Rx.Observable.create()方法可以自定義你需要的流。你需要明確通知觀察者(或者訂閱者)數(shù)據(jù)流的到達(onNext()) 或者錯誤的發(fā)生(onError())。這個實現(xiàn)中，我們封裝了jQuery 的異步 Promise。那么Promise也是可觀察對象嗎？

冰狗，你猜對啦！

可觀察對象(Observable)是超級Promise(原文Promise++，可以對比C，C++，C++在兼容C的同時引入了面向?qū)ο蟮忍匦?。在Rx環(huán)境中，你可以簡單的通過var stream = Rx.Observable.fromPromise(promise)將Promise轉(zhuǎn)換為可觀察對象，我們后面將這樣使用，唯一的區(qū)別是，可觀察對象與Promises/A+ 并不兼容, 但是理論上不會產(chǎn)生沖突。 Promise 可以看做只能發(fā)射單值的可觀察對象，Rx流則允許返回多個值。

不過，可觀察對象至少和Promise一樣強大。如果你相信針對Promise的那些吹捧，不妨也留意一下Rx環(huán)境中的可觀察對象。

回到我們的例子，細心的你肯定看到了subscribe()的嵌套使用，這和回調(diào)函數(shù)嵌套一樣令人惱火。responseStream 的確和 requestStream 存在依賴關(guān)系。前面我們不是提到過Rx有一些牛逼的工具集嗎？在Rx中我們擁有簡單的機制把一個流轉(zhuǎn)化為一個新的流，我們不妨試試。

我們先介紹 map(f)函數(shù)。該函數(shù)在流A的每個之上調(diào)用函數(shù)f() ，然后在流B上生成對應(yīng)的新值。如果在請求、響應(yīng)流上調(diào)用map(f)，我們可以將請求的URL隱射為響應(yīng)流中的Promise(此時響應(yīng)流中包含了Promise的序列)。

var responseMetastream = requestStream
  .map(function(requestUrl) {
    return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl));
  });

我們把上面代碼執(zhí)行后的返回結(jié)果稱為 metastream (譯者注：按字面可以翻譯為“元流”，即包含流的流。類似概念例如：元編程——用于生成程序的編程方法；元知識——獲取知識的知識)：包含其他流的流。沒什么嚇人的，一個metastream會在執(zhí)行后發(fā)射一個流。你可以把它看做一個指針指針)：每一個發(fā)射的值是指向另外一個流的指針。在我們的例子中，每一個URL被映射為一個指向Promise流的指針，每一個Promise流中包含了相應(yīng)的響應(yīng)信息。

(譯者注：以下給出 metastream 的方法的解析方法，方便與下面的方法進行對比)：

responseMetastream.subscribe(function(streamedPromise) {
    // 首先展開metastream，獲取內(nèi)部的流
    streamedPromise.subscribe(function(responseJsonObject) {
        // 返回內(nèi)部流發(fā)射的值
        return responseJsonObject;
    });
});

當(dāng)前版本響應(yīng)產(chǎn)生的metastream看起來有些讓人疑惑，似乎用處不大。當(dāng)前場景中，我們僅僅需要獲得簡單的響應(yīng)流，流中發(fā)射的值為簡單的JSON對象。使用flatMap:這個函數(shù)可以將枝干的流的值發(fā)射到主干流之上。當(dāng)然metastream的產(chǎn)生并不是bug，只是這個場景不適合而已，map()，flatMap()都是Rx處理異步請求工具中的一部分。(譯者注：如果流A中包含了若干其他流，在流A上調(diào)用flatMap()函數(shù)，將會發(fā)射其他流的值，并將發(fā)射的所有值組合生成新的流。)

var responseStream = requestStream
  .flatMap(function(requestUrl) {
    return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl));
  });

贊！響應(yīng)流是依照請求流定義的，如果場景中生成了更多的請求流，我們也會生成同樣多的響應(yīng)流：

請求流:  --a-----b--c------------|->
響應(yīng)流:  -----A--------B-----C---|->

(小寫字母表示請求, 大寫字母代表響應(yīng))

獲得響應(yīng)流之后，我們就可以再訂閱后渲染頁面了:

responseStream.subscribe(function(response) {
  // 在瀏覽器中渲染響應(yīng)數(shù)據(jù)的邏輯
});

馬克一下目前的代碼:

var requestStream = Rx.Observable.just("https://api.github.com/users");

var responseStream = requestStream
  .flatMap(function(requestUrl) {
    return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl));
  });

responseStream.subscribe(function(response) {
  // 在瀏覽器中渲染響應(yīng)數(shù)據(jù)的邏輯
});

刷新“你可能感興趣的用戶”

忘了說了，我們每一次請求都會返回100個GitHub用戶的數(shù)據(jù)。GitHub的API只允許我們設(shè)置頁面的偏移量但是不能設(shè)置每次獲得數(shù)據(jù)的數(shù)量。嗯，我們需要3個推薦用戶的數(shù)據(jù)，其他97個就這樣浪費了。暫時忽略這個問題，后面我們看看怎么緩存數(shù)據(jù)來減少數(shù)據(jù)的浪費。

每一次點擊刷新按鈕(高能注意：是一個按鈕，點擊后刷新“我可能感興趣的人”的數(shù)據(jù)，而不是瀏覽器的刷新按鈕)，請求流都會發(fā)射新的URL值，我們以此獲得新的響應(yīng)。刷新分為兩步：產(chǎn)生一個刷新按鈕被點擊的事件流(RP箴言：神馬都是流)；訂閱刷新事件流后改變請求流的URL地址。RxJS提供了工具方便我們將時間監(jiān)聽器轉(zhuǎn)換為可觀察對象。

var refreshButton = document.querySelector(".refresh");
var refreshClickStream = Rx.Observable.fromEvent(refreshButton, "click");

因為點擊刷新事件并不會攜帶需要請求的API的URL，我們需要把每一次點擊映射到真正的URL之上。具體實現(xiàn)方式是，在刷新點擊流發(fā)生后，我們通過產(chǎn)生隨機的頁面拼湊出URL，并向GitHub發(fā)起請求。

var requestStream = refreshClickStream
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  });

由于是簡單的教程，我并沒有寫相關(guān)的測試，但是我仍然知道原先的功能被我搞砸啦。呃。。。頁面打開后居然沒有請求流了，除非我點擊刷新按鈕，否則數(shù)據(jù)怎么都出不來。擦。。。我希望不管是點擊刷新按鈕"_還是_"第一次打開頁面，都可以產(chǎn)生獲得“我可能感興趣的人”的數(shù)據(jù)的GitHub的請求流。

把兩個流分開寫特別簡單，我們已經(jīng)知道怎么做了：

var requestOnRefreshStream = refreshClickStream
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  });
  
var startupRequestStream = Rx.Observable.just("https://api.github.com/users");

但是我們怎么把兩個流“合并”在一塊呢？使用 merge()函數(shù)吧。我們用ASCII圖表來解釋這個函數(shù)的作用：

流 A: ---a--------e-----o----->
流 B: -----B---C-----D-------->
          vvvvvvvvv merge vvvvvvvvv
          ---a-B---C--e--D--o----->

使用merge()后簡單多了:

var requestOnRefreshStream = refreshClickStream
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  });
  
var startupRequestStream = Rx.Observable.just("https://api.github.com/users");

var requestStream = Rx.Observable.merge(
  requestOnRefreshStream, startupRequestStream
);

如果不需要requestOnRefreshStream、startupRequestStream這兩個中間流，寫法更干凈、簡潔。

var requestStream = refreshClickStream
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  })
  .merge(Rx.Observable.just("https://api.github.com/users"));

還能更簡單，更有可讀性:

var requestStream = refreshClickStream
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  })
  .startWith("https://api.github.com/users");

startWith() 函數(shù)的作用和它的命名一樣。無論是什么樣的流，startWith(x) 都會把x作為這個流的啟示輸入并發(fā)射出來。上面的實現(xiàn)，還不夠DRY(Don"t repeat yourself，不要重復(fù)!)，API請求的URL地址重復(fù)了兩遍。我們將 startWith() 緊接在refreshClickStream之后，在頁面打開后就模擬一次點擊。

var requestStream = refreshClickStream.startWith("startup click")
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  });

Nice！事情不會被搞砸了，startWith()完美解決了問題。

3位“你可能感興趣的用戶”的流的構(gòu)建

目前為止，僅僅在訂閱(subscribe())時，你會觸及到“感興趣的用戶”區(qū)塊的渲染。但是通過刷新按鈕，問題接踵而至：你點擊了刷新按鈕，在新的響應(yīng)到達之前，原來的“你可能感興趣的”3個用戶并不會馬上消失。為了增強用戶體驗，我們希望在用戶點擊了刷新按鈕后就清楚老數(shù)據(jù)。

refreshClickStream.subscribe(function() {
  // 清楚舊數(shù)據(jù)： 3個你可能感興趣的用戶的DOM元素
});

停！不要用力過猛。兩個訂閱行為都會影響到這個區(qū)塊的渲染。(responseStream.subscribe()、refreshClickStream.subscribe())，并且上面的設(shè)計也不符合關(guān)注分離的理念。還記得RP 神馬都是流 的箴言嗎？

那么開始構(gòu)建這個專門的推薦流：流會發(fā)射“你可能感興趣的用戶”的JSON對象。我們會分別構(gòu)建三種這樣的流，第一種長這個樣：

var suggestion1Stream = responseStream
  .map(function(listUsers) {
    // 隨機從列表中取出一個用戶
    return listUsers[Math.floor(Math.random()*listUsers.length)];
  });

另外兩個流suggestion2Stream 和 suggestion3Stream復(fù)制粘貼就好啦。呃。。。DRY不要重復(fù)，我把這個問題作為這個教程的聯(lián)系，自己做一遍你會去思考這類場景中如何避免代碼的重復(fù)。

譯者注：如果使用UnderScore，一種方法是，新的方法總是會返回JSON Object數(shù)組:

var suggestionStream = responseStream
  .map(suggestionN(listUsers, n));

function suggestionN(listUsers, n) {
    _.times(n, function() {
        return listUsers[Math.floor(Math.random()*listUsers.length)];
    })
}

我們不再訂閱響應(yīng)流，而是變更為：

suggestion1Stream.subscribe(function(suggestion) {
  // 在區(qū)塊中渲染1位用戶的DOM元素
});

回到原始需求：“每一次刷新后，清除原來的用戶”，我們可以在刷新后，返回null作為推薦流：

var suggestion1Stream = responseStream
  .map(function(listUsers) {
    // 隨機從列表中取出一個用戶
    return listUsers[Math.floor(Math.random()*listUsers.length)];
  })
  .merge(
    refreshClickStream.map(function(){ return null; })
  );

在渲染環(huán)節(jié)，null代表無數(shù)據(jù)，我們就隱藏之前的DOM元素。

suggestion1Stream.subscribe(function(suggestion) {
  if (suggestion === null) {
    // 在區(qū)塊中隱藏一個推薦用戶的DOM元素
  }
  else {
    // 在區(qū)塊中渲染一個推薦用戶的DOM元素
  }
});

整個事件流如圖所示:

  刷新按鈕流: ----------o--------o---->
     請求流: -r--------r--------r---->
     響應(yīng)流: ----R---------R------R-->   
 推薦1個用戶: ----s-----N---s----N-s-->

N 表示 null.

頁面打開后，我們渲染“空”推薦區(qū)塊，可以通過在推薦流中附加startWith(null)實現(xiàn)：

var suggestion1Stream = responseStream
  .map(function(listUsers) {
    // 隨機從列表中取出一個用戶
    return listUsers[Math.floor(Math.random()*listUsers.length)];
  })
  .merge(
    refreshClickStream.map(function(){ return null; })
  )
  .startWith(null);

Which results in:

   刷新按鈕流: ----------o---------o---->
      請求流: -r--------r---------r---->
      響應(yīng)流: ----R----------R------R-->   
 推薦1個用戶:  -N--s-----N----s----N-s-->

關(guān)閉一個推薦元素，從緩存獲得新的推薦元素

最后一個需要實現(xiàn)的功能是：點擊"x"按鈕后關(guān)閉當(dāng)前的推薦元素，載入一個新的數(shù)據(jù)并渲染。拍腦袋意向，無論點擊了啥按鈕，我們重新請求一次新數(shù)據(jù)，生成一個新的響應(yīng)流就好了：

var close1Button = document.querySelector(".close1");
var close1ClickStream = Rx.Observable.fromEvent(close1Button, "click");
// close2Button 和 close3Button 作為練習(xí)

var requestStream = refreshClickStream.startWith("startup click")
  .merge(close1ClickStream) // 加上這個
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  });

擦，點擊了關(guān)閉按鈕整個推薦區(qū)塊都被刷新了！看來我們只有使用原來的相應(yīng)流才能解決這個bug，況且每次慷慨大方的GitHub給我們100個用戶的數(shù)據(jù)，我們只使用3個，還有1大堆留著等我們用呢，沒有必要再請求更多的數(shù)據(jù)了。

讓我們從流的角度思考，當(dāng)點擊"x"事件發(fā)生后，我們使用 最近一次的相應(yīng)流 并從中隨機取出用戶就好了：

      請求流: --r--------------->
      響應(yīng)流: ------R----------->
   點擊關(guān)閉流: ------------c----->
推薦1個用戶流: ------s-----s----->

在Rx*框架中，一個使用函數(shù)叫 combineLatest 。函數(shù)將兩個流作為輸入，并且當(dāng)其中任意一個流發(fā)射之后， combineLatest 都會組合兩個流中最新的值 a 和 b然后輸出一個新的流，流的值為 c = f(x,y) 其中 f(x, y) 是傳入的自定義函數(shù)，配合上時序圖更好理解:

流 A:     --a-----------e--------i-------->
流 B:     -----b----c--------d-------q---->
          vvvvvvvv combineLatest(f) vvvvvvv
          ----AB---AC--EC---ED--ID--IQ---->

這里的函數(shù)f，將輸入的字符串變?yōu)榇髮?/pre>
現(xiàn)在我們在 close1ClickStream 和 responseStream使用combineLatest() ， 只要用戶點擊關(guān)閉按鈕，我們就結(jié)合最新的響應(yīng)流來產(chǎn)生suggestion1Stream。 另一個方面，combineLatest() 是一個同步操作：每當(dāng)新的響應(yīng)流發(fā)射了值， 同樣會結(jié)合 close1ClickStream產(chǎn)生新的推薦數(shù)據(jù)。這樣我們大大簡化了suggestion1Stream：
var suggestion1Stream = close1ClickStream
  .combineLatest(responseStream,             
    function(click, listUsers) {
      return listUsers[Math.floor(Math.random()*listUsers.length)];
    }
  )
  .merge(
    refreshClickStream.map(function(){ return null; })
  )
  .startWith(null);
最后還有一點點問題：combineLatest()需要結(jié)合傳入的兩個流，如果其中一個流從未發(fā)射過任何值，combineLatest()將不會輸入任何新的流?；仡櫼幌律厦娴腁SCII圖表，當(dāng)?shù)谝粋€流發(fā)射值a時，不會有任何輸出，僅當(dāng)?shù)诙€流也發(fā)射了值b后，combineLatest()才會開始向外輸出。
解決方法很多，我們采取最簡單的方式(上面例子也用到過)，我們在頁面打開時限模擬一次關(guān)閉按鈕的點擊：
var suggestion1Stream = close1ClickStream.startWith("startup click") // we added this
  .combineLatest(responseStream,             
    function(click, listUsers) {
      return listUsers[Math.floor(Math.random()*listUsers.length)];
    }
  )
  .merge(
    refreshClickStream.map(function(){ return null; })
  )
  .startWith(null);
總結(jié)
再Mark一下當(dāng)前的代碼，是不是很有成就感：
var refreshButton = document.querySelector(".refresh");
var refreshClickStream = Rx.Observable.fromEvent(refreshButton, "click");

var closeButton1 = document.querySelector(".close1");
var close1ClickStream = Rx.Observable.fromEvent(closeButton1, "click");
// close2 和 close3 作為練習(xí)

var requestStream = refreshClickStream.startWith("startup click")
  .map(function() {
    var randomOffset = Math.floor(Math.random()*500);
    return "https://api.github.com/users?since=" + randomOffset;
  });

var responseStream = requestStream
  .flatMap(function (requestUrl) {
    return Rx.Observable.fromPromise($.ajax({url: requestUrl}));
  });

var suggestion1Stream = close1ClickStream.startWith("startup click")
  .combineLatest(responseStream,             
    function(click, listUsers) {
      return listUsers[Math.floor(Math.random()*listUsers.length)];
    }
  )
  .merge(
    refreshClickStream.map(function(){ return null; })
  )
  .startWith(null);
// suggestion2Stream 和 suggestion3Stream 作為練習(xí)

suggestion1Stream.subscribe(function(suggestion) {
  if (suggestion === null) {
    // 隱藏一個用戶的DOM元素
  }
  else {
    // 渲染一個新的推薦用戶的DOM元素
  }
});