原文地址：https://gist.github.com/staltz/868e7e9bc2a7b8c1f754

你所需要的一篇关于响应式编程的介绍

前言

Rxjs 是一个实现 Reactive Programming（响应式编程） 思想的库。
这篇文档通篇说的都是 Reactive Programming，有时会简称为 Reactive，在这篇文档中可以认为 Reactive Programming 就是 Rxjs。
从更广泛的意义来说，理解 Rxjs 背后的 Reactive Programming 的编程思想，反过来对于学习 Rxjs 有很大的帮助。

你应该很好奇这个叫做Reactive Programming的新事物，特别是由它衍生的包括Rx，Bacon.js，RAC 等的相关知识。

当缺乏好的学习材料时，学习 Reactive Programming 是很困难的。在我学习之初，我只找到了少部分的实用的学习指南，而且它们只介绍表面的知识，没有解决围绕它来搭建整个架构的挑战。当您试图理解某个操作符时，官方文档通常对你又没有帮助。比如说：

Rx.Observable.prototype.flatMapLatest(selector, [thisArg])

将一个可观测序列的每个元素通过合并元素的索引投射到一个新的可观测序列序列中，然后将一个可观测序列的可观测序列转换成一个仅从最近的可观测序列产生值的可观测序列。

天呐

我读了两本书，一本只描述了整体概况，而另一本则是讲如何使用 Reactive 库。最终我艰难地学习了Reactive Programming：在用它构建的同时搞清楚它。我在Futurice工作期间曾把它在一个真实的项目中，在遇到困难时得到了一些同事的帮助。

学习过程中最困难的部分就是 thinking in Reactive(响应式的思维方式)。这很大程度上是有关放弃传统编程中旧有的命令式和有状态的习惯的问题，并迫使你的大脑在不同的范式下工作。我在网上还没有找到这方面的指南，我认为这个世界应该有一个关于 thinking in Reactive 的实用教程，这样你就有一个指南了。官方文档可以在此之后为你指明道路，我希望这对你有所帮助。

什么是"Reactive Programming"？

网上有很多不好的解释和定义。维基百科和往常一样太笼统、太理论化了。 Stackoverflow的标准答案显然不适合新手。Reactive Manifesto(关于响应式的宣言？)听起来像是你向你的项目经理或推销时展示的那种东西。微软的Rx术语“Rx = Observables + LINQ + Schedulers”是如此沉重和微软，以至于我们大多数人都感到困惑。像“响应式”和“变化的传播”这样的术语并没有传达出与你的典型MV *和你最喜欢的语言之间有什么不同。当然，我的框架视图会对模型做出反应。当然，变化是会传播的。如果没有，就不会呈现任何内容。

那么，就让我们废话少说

Reactive programming是使用异步数据流进行编程。

在某种程度上，这并不是什么新鲜事。事件循环或典型的点击事件实际上是一个异步事件流，您可以对其监听并执行回调。您可以创建任何内容的数据流，而不仅仅是点击和悬停事件。流是廉价且无处不在的，任何东西都可以是流：变量，用户输入，属性，缓存，数据结构等。例如，假设您的Twitter是一个数据流，其方式与点击事件相同。您可以监听该流并做出相应的回应。
除此之外，Reactive programming还给你提供了一个功能集合，用来组合、创建和过滤这些流。这就是“函数式编程”的魔力所在。一个流可以用作另一个流的输入。甚至可以将多个流用作另一个流的输入。您可以合并两个流。您可以筛选一个流，以获得另一个流，该流只包含您感兴趣的事件。您可以将数据值从一个流映射到另一个新的流。

既然流对于响应式来说是如此重要，那么就让我们仔细研究一下它们，从我们熟悉的“点击按钮”事件流开始。

流是按时间顺序排列的一系列正在进行的事件。它可以发出三种不同的事件:值(某种类型)、错误或“完成”信号。完成事件会发生在这种情况下，我们可能会做这样的操作：关闭包含那个按钮的窗口或者视图组件。

我们只捕获发出的这些异步事件，通过定义一个函数执行回调，该函数在值发出时执行，在发出错误时执行另一个函数，在发出“完成”时执行另一个函数。有时这最后两个可以省略，您可以只专注于为值定义函数。对流的“监听”称为订阅。我们定义的函数是观察者。流是被观察的(或“可观察的”)对象。它就是是观察者模式 Observer Design Pattern。

可以将上面的示意图通过一种不同的方式来绘制数据流，这种方式叫弹珠图，我们将在本教程的某些部分使用弹珠图，如:

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

a, b, c, d 表示发出的数据
X 表示错误
|表示 '结束' 信号
---> 是时间轴

概念方面已经讲了很多，为了不让你感到无聊，下面就让我们来做一些操作:通过对原始点击事件流进行操作，生产一个新的点击事件流。
首先，我们创建一个记录按钮点击次数的事件。在常用的Reactive库中，有许多操作符可以对流进行处理，如map、filter、scan 等。当您调用其中一些操作符时，例如 clickStream.map(f)，它会返回基于点击事件流的一个新的事件流。它不会改变原始事件流。这叫做"数据不变性"(函数式的三个特性之一)的特性，它可以和响应式时间流搭配在一起使用，就像豆浆和油条一样完美的搭配。这使得我们可以通过链式调用的方式使用操作符，如clickStream.map(f).scan(g)：

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

map(f)操作符会根据f函数把原事件流中每一个返回值分别映射到新的事件流中。在上图例子中，我们把每个点击事件都映射成数字1。scan(g)操作符把之前映射的值聚集起来，通过 x = g(accumulated, current)(accumulated是累计值，current是当前值)的算法产生结果，本例中g就是一个累加函数。然后，当任一点击发生时，counterStream都将发出累计点击事件总数。

为了显示 Reactive 真正的威力，我们假设您想要一个“双击”事件流。为了让它更有趣，我们假设这个事件流同时处理"三次点击"或者"多次点击"事件。深呼吸，想象一下你将如何以一种传统的命令式和有状态的编程来实现。我敢打赌，这实现起来相当麻烦，涉及到需要定义一些变量来保存状态，还得做一些时间间隔的调整。

通过 Reactive 处理就很简单。实际上，逻辑部分只需要4行代码。但是，当前阶段先让我们忽略代码部分。无论您是初学者还是专家，通过画弹珠图来帮助思考都是理解和构建流的最佳方法。

图中，灰色方框表示将上面的事件流转换为下面的事件流的过程函数。首先我们根据点击事件250ms的间隔时间组成列表（buffer(stream.throttle, 250ms)做的事情）。现在不着急取理解实现细节，我们只关注演示 Reactive 的部分，buffer 的结果是生成一个由含有事件列表组成的流。然后使用 map() 将每个列表映射成该列表的长度传递下去。最后，使用filter(x >= 2)操作符来忽略小于1的数字。就这样通过3个操作符产生目标的数据流。我们可以通过 subscribe (订阅) 目标的数据流来验证结果是否符合我们的预期。

"我为什么要采用PR(Reactive Programming的缩写)？"

Reactive Programming 提高了代码的抽象级别，因此你可以专注于定义业务逻辑的事件之间的相互依赖，而不必不断地修改大量的实现细节。RP中的代码可能更简洁。

这种优势在现代web应用程序和移动应用程序中更为明显，这些应用程序与大量与数据事件相关的页面事件高度交互。10年前，与web页面的交互主要是向后端提交一个长表单，并向前端执行简单的呈现。应用程序已经进化得更加实时:修改单个表单字段可以自动触发保存到后端，对某些内容的“赞”可以实时反映到其他连接的用户，等等。

如今的应用程序拥有丰富的各种实时事件，为用户提供了高度交互性的体验。我们需要合适的工具来处理这个问题，而Reactive Programming 就是一个答案。

通过例子来说明 Thinking in RP

让我们来看看真正的东西。一个实际的例子，一步一步地指导如何在RP中思考。没有综合的例子，没有半解释的概念。在本教程结束时，我们将生成真正的功能代码，同时知道为什么要这样做。

我选择 JavaScript 和 RxJS 作为工具是有原因的:JavaScript是目前最让人熟悉的语言，而Rx* library系列应用在许多语言和平台中(NET、Java、Scala、Clojure、JavaScript、Ruby、Python、C++、Objective-C/Cocoa、Groovy等)。因此，无论您的工具是什么，您都可以通过学习本教程具体受益。

实现一个 "推荐关注" (Who to follow)的建议框

在Twitter中有一个页面，向你推荐可以关注的其他账户:

我们将重点模仿它的3个主要功能:

• 开始阶段，通过 API 加载推荐关注的用户账号数据，显示3个推荐用户
• 点击 "刷新按钮"，加载另外3个推荐用户显示到这三行中
• 单击每行推荐用户右上方的“x”按钮，只清除被点击的用户并显示另一个用户到当前行
• 每一行都一个用户的头像，点击可以链接到他们的主页

我们可以先忽略其他功能和按钮，因为它们是次要的。Twitter最近对未经授权的公众关闭了它的 API，我们将用 Github API 获取 users（https://api.github.com/users?since=135）代替Twitter构建我们的页面

请求与响应

在Rx中如何用处理这个问题?首先，(几乎)所有东西都可以是一条流，这就是Rx的准则。让我们从最简单的功能开始:“在开始阶段，从API加载推荐关注的用户账户数据，然后显示三个推荐用户”。这里没有什么特殊，就是 (1)发出一个请求，(2)获取响应数据，(3)渲染响应数据。我们把请求作为一个事件流。乍一看，这样做似乎有点夸张，但我们需要从基本的做起，不是吗?

开始时我们只需要做一个请求，如果我们将它作为一个数据流，它只能成为一个仅仅返回一个值的事件流而已。一会儿我们还会有很多请求要做，但当前，只有一个。

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

a是一个字符串  'https://api.github.com/users'

这是我们要请求的url事件流。无论何时发生请求事件，它都会告诉我们两件事:when 和 what。何时（when）发请求：当事件发出的时候。请求什么（what）：发出一个包含URL的字符串的值。

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

到现在，这只是一个字符串流，没有其他操作，所以我们需要在这个值被释放时做一些事情。这是通过subscribing（订阅）流来实现的。

requestStream.subscribe(function(requestUrl) {
  // 执行请求
  // onNext 改成 next,onError 改成 error, onComplete 改成 complete
  var responseStream = Rx.Observable.create(function (observer) {
     jQuery.getJSON(requestUrl)
    .done(function(response) { observer.next(response); })
    .fail(function(jqXHR, status, error) { observer.error(error); })
    .always(function() { observer.complete(); });
  });

  responseStream.subscribe(function(response) {
    // 接口响应处理
  });
}

注意到我们这里使用的是JQuery的AJAX回调方法来的处理这个异步的请求操作。但是，请稍等一下，Rx就是用来处理异步数据流的，难道它就不能处理来自请求(request)在未来某个时间响应(response)的数据流吗？好吧，理论上是可以的，让我们尝试一下。

Rx.Observable.create() 所做的是通过显式地通知每个观察者(或者说是“订阅者”)关于数据事件(next()或错误(error())，来创建自定义流。我们只是简单的封装了一下 jQuery Ajax Promise 而已。这是否意味着 jQuery Ajax Promise 本质上是一个 Observable（可观察者）呢?

是的
Observable 是 Promise++（Promise的加强版）。在Rx中，通过 var stream = Rx.Observable. frompromise (Promise) 可以很容易地将一个Promise 转换成一个Observable。唯一的不同之处在于，Observable与 Promises/A+ 不兼容，但在理论上不冲突。一个 Promise 就是一个只有一个返回值的 Observable。Rx流允许有多个值返回，更胜 Promise 一筹。

这点很棒，说明 Observable比 Promise 更强大。如果您相信 Promise 宣传的东西，可以留意一下 Rx Observable 能胜任什么。

回到我们的示例中，你应该很快注意到，我们在requestStream 的subscribe()方法中又有一个subscribe()调用，这有点类似于回调地狱。除此之外，responseStream的创建依赖于requestStream。我们之前说过，在Rx中，有很多简单的机制从其他事件流转换并创建出一个新的流，所以我们也可以这样做试试。

你现在需要了解的一个基本的操作符是 map(f),它可以从事件流A中取出每个值,对每个值执行f()函数，然后将产生的新值填充到事件流B中。如果将它应用到我们的请求和响应事件流当中,我们可以将请求 URLs 映射到响应 Promises上(伪装成流)。

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

然后我们就创建了一个叫做“metastream”(高阶流)的野兽:装载了事件流的事件流。不要惊慌。metastream也是一个流，其中每个发出的值又是另一个事件流。您可以将它看作指针数组:每一个单独发出的值就是一个指针，它指向另一个事件流。在我们的示例里，每一个请求URL都映射到一个指向包含响应数据的promise数据流。

一个响应的metastream的，看起来确实令人困惑，对我们似乎也没什么帮助。我们只想要一个简单的响应数据流，其每个发出的值都是JSON对象，而不是一个“Promise”的JSON对象。让我们来见识一下另一个函数Flatmap:map()操作符的一个版本，它通过将在“分支”流上发出的所有内容发到“主干”流来“打平”metastream。Flatmap不是“修复”metastream，metastream也不是一个bug，它们实际上是用于处理Rx中的异步响应的好工具。

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

由于响应流是根据请求流定义的，如果以后有更多的事件发生在请求流上，我们就会有相应的响应事件发生在响应流上，正如我们所期望的:

requestStream:  --a-----b--c------------|->
responseStream: -----A--------B-----C---|->

(每个小写字母都是一个请求, 大写字母是对应的响应)

现在我们终于有了一个响应流，并且可以用我们收到的数据来渲染了：

responseStream.subscribe(function(response) {
  // 以你想要的方式将 response 响应熏染到 DOM 中
});

让我们把所有代码合起来，看一下：

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) {
  // 以你想要的方式将 response 响应熏染到 DOM 中
});

刷新按钮

我还没有提到响应中的 JSON 是一个包含100个用户数据的列表。API只允许我们指定页面offset，而不允许指定页面大小，因此我们只使用了3条数据对象，而浪费了97个其他对象。现在我们可以先忽略这个问题，稍后我们将学习如何缓存响应的数据。

每次单击refresh按钮时，请求流应该发出一个新的URL，以便我们能够获取新的响应数据。我们需要两件东西:刷新按钮上的事件流(准则:一切都可以作为流)，我们需要将点击刷新按钮的事件流作为请求事件流的依赖(即点击刷新事件流会引起请求事件流)。幸运的是，RxJS 中有将事件监听器转换成 Observables 的方法了。

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

由于刷新按钮的点击事件本身不携带任何API URL，所以我们需要将每次点击映射到一个实际的URL。现在，我们将请求流更改为刷新按钮的点击事件流，该流每次都映射到API端点，并带有一个随机偏移参数。

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

因为我比较笨而且也没有使用自动化测试，所以我刚把之前做好的一个功能搞烂了。这样，请求在一开始的时候就不会执行，而只有在点击事件发生时才会执行。我们需要的是两种情况都要执行：刚开始打开网页和点击刷新按钮都会执行的请求。

我们知道如何为每一种情况做一个单独的事件流:

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()。在弹珠图中解释，它的功能如下:

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

现在应该很容易:

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

有一种替代的更简洁的写法，不需要中间流。

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()操作符如你预期的执行了操作。无论您的输入流看起来如何，startWith(x)的输出流在开始时都会有一个x作为开头。但是我没有总是DRY(Don't_repeat_yourself)，我在重复API的 URL字符串。改进这个问题的方法是将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;
 });

var responseStream = refreshClickStream
     .mergeMap((requestUrl)=>{
       return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl));
     });

好了。如果回到我“破坏自动化测试”的地方，您应该会看到对比两个地方的唯一区别是我添加了startWith()。

用事件流将3个推荐的用户数据模型化

到目前为止，在responseStream的subscribe()发生的渲染步骤中，我们只稍微提及了一下推荐关注页面。现在有了刷新按钮，我们会出现一个问题:当你点击刷新按钮，当前的三个推荐关注用户没有被清除，而只要响应的数据达到后我们就拿到了新的推荐关注的用户数据。为了让UI看起来更漂亮，我们需要在点击刷新按钮的事件发生时清除当前的三个推荐用户。

refreshClickStream.subscribe(function() {
  // 清除这三个推荐的 DOM 元素
});

不，老兄，还没那么快。我们又出现了新的问题，因为我们现在有两个订阅者影响着推荐用户的 UI DOM元素(refreshClickStream.subscribe()和responseStream.subscribe())，这听起来不符合 Separation of concerns(关注点分离)。还记得Reactive 的准则吗?

因此，让我们把推荐关注的用户数据模型化成事件流形式，每个被发出的值是一个包含推荐关注用户数据的JSON对象。我们将对这3个用户数据分开处理。下面是推荐关注的1号用户数据的事件流：

var suggestion1Stream = responseStream
  .map(function(listUsers) {
    // 从 listUsers 中获取一个随机用户
    return listUsers[Math.floor(Math.random()*listUsers.length)];
  });

其他的，如推荐关注的2号用户数据的事件流suggestion2Stream和推荐关注的3号用户数据的事件流suggestion3Stream 都可以方便的从suggestion1Stream 复制粘贴就好。这里并不是重复代码，只是为让我们的示例更加简单，而且我认为这是一个思考如何避免重复代码的好案例。

我们不在responseStream的subscribe()中处理渲染了，我们在这里这样做:

suggestion1Stream.subscribe(function(suggestion) {
  // 将1号推荐用户渲染到 DOM中
});

回到“当刷新时，清除掉当前的推荐关注的用户”，我们可以简单地将刷新按钮点击映射用户推荐数据为空null，并且在suggestion1Stream中包含进来，如下所示:

var suggestion1Stream = responseStream
  .map(function(listUsers) {
    // 从 listUsers 中获取一个随机的 user
    return listUsers[Math.floor(Math.random()*listUsers.length)];
  })
  .merge(
    refreshClickStream.map(function(){ return null; })
  );

在渲染时，我们将null解释为“没有数据”，然后把页面元素隐藏起来。

suggestion1Stream.subscribe(function(suggestion) {
  if (suggestion === null) {
    // 隐藏第一个推荐 DOM 元素
  }
  else {
    // 显示第一个推荐 DOM 元素，渲染数据
  }
});

现在我们的一个大的示意图是这样的：

refreshClickStream: ----------o--------o---->
     requestStream: -r--------r--------r---->
    responseStream: ----R---------R------R-->   
 suggestion1Stream: ----s-----N---s----N-s-->
 suggestion2Stream: ----q-----N---q----N-q-->
 suggestion3Stream: ----t-----N---t----N-t-->

N代表null。
作为一种补充，我们可以在一开始的时候就渲染空的推荐内容。这通过把startWith(null)添加到推荐关注的事件流就可以了：

var suggestion1Stream = responseStream
  .map(function(listUsers) {
    // 从 listUsers 中获取一个随机的 user
    return listUsers[Math.floor(Math.random()*listUsers.length)];
  })
  .merge(
    refreshClickStream.map(function(){ return null; })
  )
  .startWith(null);

结果为:

refreshClickStream: ----------o---------o---->
     requestStream: -r--------r---------r---->
    responseStream: ----R----------R------R-->   
 suggestion1Stream: -N--s-----N----s----N-s-->
 suggestion2Stream: -N--q-----N----q----N-q-->
 suggestion3Stream: -N--t-----N----t----N-t-->

推荐关注的关闭和使用已缓存的响应数据

只剩这一个功能没有实现了，每个推荐关注的用户UI会有一个'x'按钮来关闭自己，然后在当前的用户数据UI中加载另一个推荐关注的用户。最初的想法是：点击任何关闭按钮时都需要发起一个新的请求：

var close1Button = document.querySelector('.close1');
var close1ClickStream = Rx.Observable.fromEvent(close1Button, 'click');
// 第二三个推荐用户的关闭按钮也是一样的操作

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

这样没什么效果，这样会关闭和重新加载全部的推荐关注用户，而不仅仅是处理我们点击的那一个。这里有几种方式来解决这个问题，并且让它变得有趣，我们将重用之前的请求数据来解决这个问题。这个API响应的每页数据大小是100个用户数据，而我们只使用了其中三个，所以还有一大堆未使用的数据可以拿来用，不用去请求更多数据了。

接下来，我们继续用事件流的方式来思考。当'close1'点击事件发生时，我们想要使用最近发出的响应数据，并执行responseStream函数来从响应列表里随机的抽出一个用户数据来，就像下面这样：

    requestStream: --r--------------->
   responseStream: ------R----------->
close1ClickStream: ------------c----->
suggestion1Stream: ------s-----s----->

在 Rx* 中有一个组合操作符叫 combineLatest，它似乎能够实现我们的需求。它接受两个数据流A和B作为输入，并且无论哪一个数据流发出一个值了，combineLatest就将从两个数据流最近发出的值a和b作为f函数的输入，计算后返回一个输出值（c = f(x,y)）。用弹珠图会让这个函数的过程看起来比较好理解:

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

f 是将字符转大写字母的函数

这样我们可以在close1ClickStream和responseStream上应用combineLatest()，只要点击close1按钮，我们就可以获得最近的响应数据，并在suggestion1Stream上产生一个新值。另一方面，combineLatest()也是相对的:每当在responseStream上发出一个新的响应时，它将会结合一个新的点击关闭按钮事件来产生一个新的推荐关注的用户数据。这很有趣，因为它可以给我们简化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()使用两个最近的数据源，但是如果其中一个源还没有发出任何东西，combineLatest()就不能在输出流上生成数据事件。如果您查看上面的弹珠图，您会看到当第一个流发出值a时，输出为空。只有当第二个流发出值b时，才会产生输出值。
这里有很多种方法来解决这个问题，我们使用最简单的一种，也就是在启动的时候模拟'close 1'的点击事件：

var suggestion1Stream = close1ClickStream.startWith('startup click') // 添加这行代码
  .combineLatest(responseStream,             
    function(click, listUsers) {l
      return listUsers[Math.floor(Math.random()*listUsers.length)];
    }
  )
  .merge(
    refreshClickStream.map(function(){ return null; })
  )
  .startWith(null);

封装

我们已经完成了。下面是封装好的完整示例代码：

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 按钮也是一样的逻辑

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 推荐流也是一样的处理逻辑

suggestion1Stream.subscribe(function(suggestion) {
  if (suggestion === null) {
    // 隐藏第一个推荐 DOM 元素
  }
  else {
    // 显示第一个推荐 DOM 元素，渲染数据
});

您可以看到可演示的示例工程:http://jsfiddle.net/staltz/8jFJH/48/

以上的代码片段虽小但做到很多事:它适当地使用关注分离原则(separation of concerns)实现了对多个事件流的管理，甚至做到了响应数据的缓存。这种函数式的风格使得代码看起来更像是声明式编程而非命令式编程，我们并不是在给一组指令去执行，只是定义了事件流之间关系来告诉它这是什么。例如，在Rx中我们告诉计算机suggtion1stream是suggtion1stream是'close 1'事件结合从最新的响应数据中拿到的一个用户数据的数据流，除此之外，当刷新事件发生时和程序启动时，它就是null。

留意一下代码中并未出现例如if、for、while等流程控制语句，或者像JavaScript程序中典型的基于回调的流程控制。如果可以的话，你甚至可以在subscribe()上使用filter()来开拓if和else(我将把实现细节留给您作为练习)。在Rx中，我们有诸如map、filter、scan、merge、combineLatest、startWith等数据流操作符。还有很多函数可以用来控制事件驱动编程的流程。这些函数的集合可以让你使用更少的代码实现更强大的功能。

接下来

如果您认为 Rx* 将是您首选的 Reactive Programming 库，那么请花些时间来熟悉big list of functions（操作符列表）用来转换、组合和创建 Observables。如果您想通过弹珠图了解这些操作符，请查看 RxJava's very useful documentation with marble diagrams。无论何时你遇到问题，画出那些弹珠图，思考一下，看看一大串操作符，然后继续思考。根据我的经验，这样效果很有效。

一旦您开始使用 Rx* 编程的窍门，就必须要理解Cold 与 Hot Observables的概念。如果你忽视这一点，它会回过头狠狠地咬你一口。我这里已经警告你了，学习函数式编程，并熟悉影响Rx*的副作用等问题，将进一步提升您的技能。

但是 Reactive Programming 不仅仅是 Rx。像Bacon.js，它使用起来很直观，没有您在 Rx 中有时会遇到的怪癖。Elm 语言则以它自己的方式支持响应式编程:它是一种可以编译为JavaScript + HTML + CSS的 Reactive Programming，并且有一个time travelling debugger功能，非常棒。

Rx 适用于事件较多的前端和应用程序。但这不仅仅是客户端问题，还可以用在后端或者接近数据库的地方。事实上，RxJava是Netflix 服务端 API 用来处理并行的的组件。Rx不局限于一种特定类型的应用程序或语言。它真的是你编写任何事件驱动程序、可以遵循的一个非常棒的编程范式。

如果本教程对您有帮助，请转发它。