RxJs 介绍 rxjs教程

"Reactive Programming是神马？"

互联网上充斥着很多操蛋的解释。维基百科又宽泛又玄乎。Stackoverflow教科书式的解释非常不适合信任Reactive Manifesto听起来像是给给项目经理或者是销售的汇报。 微软的Rx 定义"Rx = Observables + LINQ + Schedulers" 太重并且太微软化了，让人看起来不知所云。“响应”、“变化发生”这些术语无法很好地阐释Reactive Programming的显著特点，听起来和你熟悉的MV*、编程语言差别不大。 当然，我的视角也是基于模型和变换的，要是脱离了这些概念，一切都是无稽之谈了。

那么我要开始吧啦吧啦了。

Reactive programming 是针对异步数据流的编程。

一定程度而言，Reactive programming并不算新的概念。Event Bus、点击事件都是异步流。开发者可以观测这些异步流，并调用特定的逻辑对它们进行处理。使用Reactive如同开挂：你可以创建点击、悬停之类的任意流。通常流廉价（点击一下就出来一个）而无处不在，种类丰富多样：变量，用户输入，属性，缓存，数据结构等等都可以产生流。举例来说：微博回文（译者注：比如你关注的微博更新了）和点击事件都是流：你可以监听流并调用特定的逻辑对它们进行处理。

基于流的概念，Reactive赋予了你一系列神奇的函数工具集，使用他们可以合并、创建、过滤这些流。一个流或者一系列流可以作为另一个流的输入。你可以_合并_

两个流，从一堆流中_过滤_你真正感兴趣的那一些，将值从一个流_映射_到另一个流。

如果流是Reactive programming的核心，我们不妨从“点击页面中的按钮”这个熟悉的场景详细地了解它。

流是包含了有时序，正在进行事件的序列，可以发射(emmit)值（某种类型）、错误、完成信号。流在包含按钮的浏览器窗口被关闭时发出完成信号。

我们异步地捕获发射的事件，定义一系列函数在值被发射后，在错误被发射后，在完成信号被发射后执行。有时，我们忽略对错误，完成信号地处理，仅仅关注对值的处理。对流进行监听，通常称为订阅，处理流的函数是观测者，流是被观测的主体。这就是观测者设计模式。

教程中，我们有时会使用ASCII字符来绘制图表：

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

a, b, c, d 是数据流发射的值
X 是数据流发射的错误
| 是完成信号
---> 是时序轴

哔哔完了，我们来点新的，不然很快你就感觉到寂寞了。我们将把原来的点击事件流转换为新的点击事件流。

首先我们创建一个计数流来表明按钮被点击的次数。在Reactive中，每一个流都拥有一些列方法，例如map，filter，scan等等。当你在流上调用这些方法，例如clickStream.map(f)，会返回基于点击事件流的新的流，同时原来的点击事件流并不会被改变，这个特性被称为不可变性（immutability）。不可变性与Reactive配合相得益彰，如同美酒加咖啡。我们可以链式地调用他们：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)函数对原来的流使用我们出入的f函数进行转换，并生成新的流。在上面的例子中，我们将每一次点击映射为数字1。scan(g)函数将所有流产生的值进行汇总，通过传入x = g(accumulated, current)函数产生新的值，g是简单的求和函数。最后counterStream在点击发生后发射点击事件发生的总数。

为了展示Reactive的真正力量，我们举个例子：你想要“两次点击”事件的流，或者是“三次点击”，或者是n次点击的流。深呼吸一下，试着想想怎么用传统的命令、状态式方法来解决。我打赌这个这会相当操蛋，你会搞些变量来记录状态，还要搞些处理时延的机制。

如果用Reactive来解决，太他妈简单了。实际上4行代码就可以搞定。先不要看代码，不管你是菜鸟还是牛逼，使用图表来思考可以使你更好地理解构建这些流的方法。

灰色框里面的函数会把一个流转换成另外一个流。首先我们把点击打包到list中，如果点击后消停了250毫秒，我们就重新打包一个新的list（显然buffer(stream.throttle(250ms))就是用来干这个的，不明白细节没有关系，反正是demo嘛）。我们在列表上调用map，将列表的长度映射为一个整数的流。最后，我们通过filter(x >= 2)过滤掉整数1。哈哈：3个操作就生成了我们需要的流，现在我们可以订阅（监听）这个流，然后来完成我们需要的逻辑了。

通过这个例子，我希望你能感受到使用Reactive的牛逼之处了。这仅仅是冰山一角。你可以在不同地流上（比如API响应的流）进行同样的操作。同时，Reactive还提供了许多其他实用的函数。

"我要在今后的项目中使用Reactive programming吗？"

Reactive Programming 提高了编码的抽象程度，你可以更好地关注在商业逻辑中各种事件的联系避免大量细节而琐碎的实现，使得编码更加简洁。

使用Reactive Programming，将使得数据、交互错综复杂的web、移动app开发收益更多。10年以前，与网页的交互仅仅是提交表单、然后根据服务器简单地渲染返回结果这些事情。App进化得越来越有实时性：修改表单中一个域可以同步地更新到后端服务器。“点赞”信息实时地在不同用户设备上同步。

现代App中大量的实时事件创造了更好的交互和用户体验，披荆斩棘需要利剑在手，Reactive Programming就是你手中的利剑。

Reactive Programming编程思想（附实例）

我们将从实例可以深入Reactive Programming的编程思想，文章末尾，一个完整地实例应用会被构建，你也会理解整个过程。

我选择 JavaScript和RxJS作为构建的基础, 大多开发者都熟悉JavaScript语言。Rx* library family在各种语言和平台都是实现 (.NET,Java,Scala,Clojure,JavaScript,Ruby,Python,C++,Objective-C/Cocoa,Groovy, 等等)。无论你选择在哪个平台或者那种语言实践Reactive Programming，你都将从本教程中受益。

微博（Twitter）简易版“你可能感兴趣的人”

微博主页，有一个组件会推荐给你那些你可能感兴趣的人。

我们的Demo将使用这个场景，关注下面这些主要特性：

• 页面打开后，通过API加载数据展示3个你可能感兴趣的用户账号

• 点击“刷新”按钮，重新加载三个新的用户账号

• 在一个用户账号上点击'x' 按钮，清除当前这个账户，重新加载一个新的账户

• 每行展示账户的信息和这个账户主页的链接

其他特性和按钮我们暂且忽略，由于Twitter在最近关闭了公共API授权接口，我们选择Github作为代替，展示GitHub用户的账户。实例中我们使用该接口获取GitHub用户.

如果你希望先睹为快，完成后的代码已经发布在了Jsfiddle。

请求&响应

这个问题使用Rx怎么解?，呵呵，我们从Rx的箴言开始：_神马都是流_。首先我们做最简单的部分——页面打开后通过API加载3个账户的信息。分三步走：(1)发一个请求(2)获得响应(3)依据响应渲染页面。那么，我们先使用流来表示请求。我靠，表示个请求用得着吗？不过千里之行始于足下。

页面加载时，仅需要一个请求。所以这个数据流只包含一个简单的反射值。稍后，我们再研究如何多个请求出现的情况，现在先从一个请求开始。

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

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

这个流中包含了我们希望请求的URL地址。一旦这个请求事件发生，我们可以获知两件事情：请求流发射值（字符串URL）的时间就是请求需要被执行的时间，请求需要请求的地址就是请求流发射的值。

在Rx*中构建一个单值的流很容易。官方术语中把流称为“观察的对象”（"Observable"），因为流可以被观察、订阅，这么称呼显得很蠢，我自己把他们称为_stream_。

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

目前这个携带字符串的流没有其他操作,我们需要在这个流发射值之后，做点什么：通过订阅这个流来实现。

requestStream.subscribe(function(requestUrl) {
  // 执行异步请求
  jQuery.getJSON(requestUrl, function(responseData) {
    // ...
  });
}

我们采用了jQuery的Ajax回调 (假设读着已经了解jQuery ajax回调) 来处理异步请求操作。 且慢，Rx天生就是处理异步数据流的，

为何不把请求的响应作为一个携带数据的流呢？ 么么哒，概念上没有问题，我们就来操作一下。

requestStream.subscribe(function(requestUrl) {
  // 执行异步请求
  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) {
    // 业务逻辑
  });
}

使用Rx.Observable.create方法可以自定义你需要的流。你需要明确通知观察者（或者订阅者）数据流的到达(onNext) 或者错误的发生(onError)。这个实现中，我们封装了jQuery 的异步 Promise。那么Promise也是可观察对象吗？

冰狗，你猜对啦！

可观察对象（Observable）是超级Promise（原文Promise++，可以对比C，C++，C++在兼容C的同时引入了面向对象等特性）。 在Rx环境中，你可以简单的通过var stream = Rx.Observable.fromPromise(promise)将Promise转换为可观察对象， 我们后面将这样使用， 唯一的区别是，可观察对象与Promises/A+并不兼容, 但是理论上不会产生冲突。 Promise 可以看做只能发射单值的可观察对象，Rx流则允许返回多个值。

不过，可观察对象至少和Promise一样强大。如果你相信针对Promise的那些吹捧，不妨也留意一下Rx环境中的可观察对象。

回到我们的例子，细心的你肯定看到了subscribe的嵌套使用，这和回调函数嵌套一样令人恼火。responseStream的确和requestStream存在依赖关系。前面我们不是提到过Rx有一些牛逼的工具集吗？在Rx中我们拥有简单的机制把一个流转化为一个新的流，我们不妨试试。

我们先介绍 map(f)函数。该函数在流A的每个之上调用函数f， 然后在流B上生成对应的新值。如果在请求、响应流上调用map(f)，我们可以将请求的URL隐射为响应流中的Promise（此时响应流中包含了Promise的序列）。

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

我们把上面代码执行后的返回结果称为 "_metastream_"（译者注：按字面可以翻译为“元流”，即包含流的流。类似概念例如：元编程——用于生成程序的编程方法；元知识——获取知识的知识）：包含其他流的流。没什么吓人的， 一个metastream会在执行后发射一个流。 你可以把它看做一个指针 指针)： 每一个发射的值是指向另外一个流的指针。在我们的例子中，每一个URL被映射为一个指向Promise流的指针，每一个Promise流中包含了相应的响应信息。

（译者注：以下给出"_metastream_"的方法的解析方法，方便与下面的方法进行对比）：

responseMetastream.subscribe(function(streamedPromise) {
    // 首先展开metastream，获取内部的流
    streamedPromise.subscribe(function(responseJsonObject) {
        // 返回内部流发射的值
        return responseJsonObject;
    });
});

当前版本响应产生的metastream看起来有些让人疑惑，似乎用处不大。当前场景中，我们仅仅需要获得简单的响应流，流中发射的值为简单的JSON对象。使用flatMap:这个函数可以将枝干的流的值发射到主干流之上。当然metastream的产生并不是bug，只是这个场景不适合而已，map，flatMap都是Rx处理异步请求工具中的一部分。（译者注：如果流A中包含了若干其他流，在流A上调用flatMap()函数，将会发射其他流的值，并将发射的所有值组合生成新的流。）

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

赞！响应流是依照请求流定义的，如果场景中生成了更多的请求流，我们也会生成同样多的响应流：

请求流:  --a-----b--c------------|->
响应流:  -----A--------B-----C---|->

(小写字母表示请求, 大写字母代表响应)

获得响应流之后，我们就可以再订阅后渲染页面了:

responseStream.subscribe(function(response) {
  // 在浏览器中渲染响应数据的逻辑
});

Mark一下，目前我们的代码写了这些:

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) {
  // 在浏览器中渲染响应数据的逻辑
});