您一直错过的反应式编程入门

如果您更喜欢观看实况编码的视频教程，请查看我用与本文相同的内容录制的这个系列：Egghead.io-反应式编程简介。

所以您很好奇学习这个叫做反应式编程的新东西，特别是它的变体，包括Rx、Bacon.js、RAC和其他。

学习它很难，由于缺乏好的材料，学习起来就更难了。刚开始的时候，我试着找教程。我只找到了几个实用指南，但他们只是触及了皮毛，从来没有解决过围绕它构建整个体系结构的挑战。当您试图理解某些功能时，库文档通常无济于事。我是说，老实说，看看这个：

通过合并元素的索引将可观测序列的每个元素投影到新的可观测序列序列中，然后将可观测序列的可观测序列变换为仅产生来自最新可观测序列的值的可观测序列。

我已经读了两本书，一本只是描绘了大局，而另一本则潜心研究如何使用反应式图书馆。我最终以一种艰难的方式学习了反应式编程：一边用它构建，一边算出它。在我在Futurice的工作中，我在一个真正的项目中使用了它，当我遇到麻烦时，我得到了一些同事的支持。

学习之旅中最艰难的部分是被动思考。这在很大程度上是关于放弃典型编程的旧的命令性和有状态的习惯，迫使你的大脑在不同的范例中工作。我在互联网上还没有找到任何关于这方面的指南，我认为这个世界应该有一个关于如何被动思考的实用教程，这样你就可以开始了。在此之后，库文档可以为您指明方向。我希望这对你有帮助。

互联网上有很多糟糕的解释和定义。维基百科一如既往地过于笼统和理论化。StackOverflow的标准答案显然不适合新来者。反应性宣言听起来像是你向你的项目经理或你公司的商人展示的那种东西。微软的处方术语(Rx=Observables+LINQ+Schedulers&34；)是如此繁琐和繁琐，以至于我们大多数人都感到困惑。像“被动”和“传播变化”这样的术语并没有传达出与你的典型MV*和你最喜欢的语言已经做的有什么特别的不同。当然，我的框架视图会对模型做出反应。当然，变化是会传播的。如果它不这样做，什么都不会呈现。

在某种程度上，这并不是什么新鲜事。事件总线或典型的单击事件实际上是异步事件流，您可以在其上观察和执行一些副作用。反应性是类固醇的想法。您可以创建任何内容的数据流，而不仅仅是从单击和悬停事件创建数据流。流既便宜又无处不在，任何东西都可以是流：变量、用户输入、属性、缓存、数据结构等。例如，假设您的Twitter提要是一个数据流，其方式与点击事件相同。您可以监听该流并做出相应的反应。

最重要的是，您可以使用一个令人惊叹的工具箱来组合、创建和过滤这些流。这就是功能性魔法发挥作用的地方。一个流可以用作另一个流的输入。即使是多个流也可以用作另一个流的输入。您可以合并两个流。您可以筛选一个流，以获得只包含您感兴趣的那些事件的另一个流。您可以将数据值从一个流映射到另一个新流。

如果流是反应性的核心，那么让我们从我们熟悉的单击按钮事件流开始，仔细研究一下它们吧。

流是按时间排序的一系列正在进行的事件。它可以发出三种不同的东西：值(某种类型)、错误或已完成信号。例如，假设在关闭包含该按钮的当前窗口或视图时发生"；Complete"；。

我们只异步捕获这些发出的事件，方法是定义一个值发出时将执行的函数、发出错误时执行的另一个函数和发出'；完成时发出的另一个函数。有时后两个可以省略，您只需专注于定义值的函数即可。监听该流的称为订阅。我们定义的函数是观察者。流是被观察的对象(或可观察的)。这正是观察者设计模式。

绘制该图表的另一种方法是使用ASCII，我们将在本教程的某些部分使用该方法：

--a-b-c-d-X-|-->；a、b、c、d是发射值X是错误|是'；完成'；信号->；是时间线。

既然这感觉已经很熟悉了，而且我不想让您感到厌烦，让我们来做点新的事情：我们将从原来的单击事件流转换成新的单击事件流。

首先，让我们创建一个计数器流来指示按钮被点击了多少次。在常见的反应库中，每个流都附加了许多函数，如map、filter、scan等。当您调用其中一个函数(如clickStream.map(F))时，它将根据单击流返回一个新流。它不会以任何方式修改原始点击流。这是一种被称为不变性的属性，它与反应性流一起使用，就像煎饼和糖浆一起吃一样。这允许我们链接诸如clickStream.map(F).scan(G)这样的函数：

Map(F)函数根据您提供的函数f替换(到新流中)每个发射值。在我们的示例中，我们在每次单击时都映射到数字1。Scan(G)函数聚合流上所有先前的值，产生值x=g(累积的，当前的)，其中g在本例中只是加法函数。然后，每当发生单击时，Counterstream都会发出总的单击次数。

为了展示反应性的真正威力，让我们只说你想要有一系列的双击事件吧。为了让它更有趣，让我们假设我们希望新的流将三次点击视为双击，或者通常认为是多次点击(两次或更多)。深吸一口气，想象一下你将如何以传统的命令和有状态的方式做到这一点。我敢打赌，这听起来相当糟糕，而且涉及到一些保持状态的变量和一些摆弄时间间隔的因素。

嗯，在反应性方面，这是相当简单的。实际上，逻辑只有4行代码，但是让我们暂时忽略代码。无论您是初学者还是专家，用图表思考都是理解和构建流的最好方式。

灰盒是将一个流转换为另一个流的函数。首先，我们在列表中累积点击，只要发生250毫秒的事件静默(简而言之，这就是Buffer(Stream.throttle(250ms))所做的事情)。不要担心在这一点上理解细节，我们现在只是演示反应性)。结果是一个列表流，我们从其中应用map()将每个列表映射到一个与该列表的长度匹配的整数。最后，我们使用FILTER(x>；=2)函数忽略1个整数。这就是：生产我们想要的流的3个操作。然后我们就可以订阅(收听)它来做出我们想要的相应反应。

我希望你喜欢这种方法的美感。这个示例只是冰山一角：您可以对不同类型的流应用相同的操作，例如，对API响应流应用相同的操作；另一方面，还有许多其他函数可用。

反应式编程提高了代码的抽象级别，这样您就可以专注于定义业务逻辑的事件之间的相互依赖，而不必经常纠结于大量的实现细节。RP中的代码可能会更简洁。

这一好处在与大量与数据事件相关的UI事件高度互动的现代Web应用程序和移动应用程序中更为明显。10年前，与网页的交互基本上是向后端提交一个很长的表单，然后向前端执行简单的渲染。应用程序已经向更实时的方向发展：修改单个表单域可以自动触发保存到后端，对某些内容的点赞可以实时反映给其他连接的用户，以此类推。

如今的应用程序拥有丰富的各种实时事件，为用户提供了高度互动的体验。我们需要工具来恰当地处理这一问题，而反应式编程就是答案。

让我们一头扎进真正的东西里吧。这是一个真实世界的例子，有一个关于如何用RP思考的循序渐进的指南。没有人工合成的例子，没有半解的概念。在本教程结束时，我们将生成真正有效的代码，同时了解我们为什么要做每件事。

我选择JavaScript和RxJS作为工具是因为：JavaScript是目前最熟悉的语言，Rx*库家族广泛适用于多种语言和平台(.NET、Java、Scala、Clojure、JavaScript、Ruby、Python、C++、Objective-C/Cocoa、Groovy等)。因此，无论您的工具是什么，您都可以通过遵循本教程获得具体的好处。

单击帐户行上的按钮时，仅清除该当前帐户并显示另一个帐户。

我们可以省略其他功能和按钮，因为它们是次要的。而且，与其最近关闭其API对未经授权的公众开放的Twitter，不如让我们为在Github上关注用户构建用户界面。有一个Github API可以用来获取用户。

这方面的完整代码已经在http://jsfiddle.net/staltz/8jFJH/48/上准备好了，如果您想要达到峰值的话。

您如何使用Rx解决此问题？嗯，首先，(几乎)一切都可以是溪流。这就是处方咒语。让我们从最简单的功能开始：在启动时，从API加载3个帐户数据。这里没有什么特别的，这只是(1)执行请求，(2)获取响应，(3)呈现响应。因此，让我们继续并将我们的请求表示为流。乍一看，这感觉有点矫枉过正，但我们需要从基础做起，对吗？

在启动时，我们只需要执行一个请求，所以如果我们将其建模为数据流，那么它将是一个只有一个发送值的流。稍后，我们知道会有许多请求发生，但目前，这只是一个请求。

这是我们要请求的URL流。每当请求事件发生时，它都会告诉我们两件事：时间和内容。"；何时"；应该执行请求的时间是事件发出时。并且"；应该请求的是发出的值：包含URL的字符串。

在Rx*中，使用单个值创建这样的流非常简单。流的官方术语是"；Observable"；，因为它可以被观察到，但我觉得它是一个愚蠢的名字，所以我称它为流。

但是现在，这只是一个字符串流，不做其他操作，所以我们需要在发出该值时以某种方式使某些事情发生。这是通过订阅流来实现的。

注意，我们使用jQuery Ajax回调(假设您已经知道)来处理请求操作的异步性。但请稍等片刻，Rx用于处理异步数据流。该请求的响应不能是包含将来某个时间到达的数据的流吗？嗯，在概念层面上，它看起来确实很像，所以让我们试一试吧。

RequestStream。SUBSCRIBE(function(RequestUrl){//执行请求var responseStream=Rx。看得见。CREATE(函数(观察者){jQuery.。GetJSON(RequestUrl)。完成(函数(响应){观察者。OnNext(响应)；})。FAIL(Function(jqXHR，Status，Error){观察者。OnError(Error)；})。Always(Function(){观察者。OnCompleted()；})；})；ResponseStream。Subscribe(function(Response){//对响应做点什么})；}。

Rx.Observable.create()所做的是通过显式通知每个观察者(或者换句话说，订阅者)有关数据事件(onNext())或错误(onError())的数据事件(onNext())或错误(onError())来创建您自己的定制流。我们所做的只是包装jQuery Ajax承诺。打扰一下，这是不是意味着承诺就是可遵守的？

可观察的是Promise++。在Rx中，您可以通过执行var stream=Rx.Observable.FromPromise(Promise)轻松地将承诺转换为可观察对象，所以让我们使用它。唯一的区别是，可观测对象不符合Promises/A+，但在概念上没有冲突。承诺只是一个具有单一发射值的可观察物。RX流允许许多返回值，超出了承诺。

这是相当不错的，并且显示了可观测性至少和承诺一样强大。因此，如果你相信承诺的炒作，请密切关注Rx观察者的能力。

现在回到我们的示例，如果您很快注意到的话，我们在另一个调用中有一个Subscribe()调用，这有点类似于回调地狱。此外，ResponseStream的创建依赖于requestStream。正如您以前听到的，在Rx中有一些简单的机制可以转换其他流并创建新的流，所以我们应该这样做。

到目前为止，您应该知道的一个基本函数是map(F)，它获取流A的每个值，对其应用f()，然后在流B上生成值。如果我们对请求流和响应流执行此操作，则可以将请求URL映射到响应承诺(伪装成流)。

然后，我们将创建一个名为#34；metastream"；的野兽：一条溪流。先别惊慌失措。传输流是一个流，其中每个发射值都是另一个流。您可以将其视为指针：每个发出的值都是指向另一个流的指针。在我们的示例中，每个请求URL被映射到一个指向包含相应响应的承诺流的指针。

响应的转移流看起来令人困惑，而且似乎对我们一点帮助都没有。我们只需要一个简单的响应流，其中发出的每个值都是一个JSON对象，而不是JSON对象的承诺。向Flatmap先生问好：MAP()的一个版本，它通过在主干上发射将在分支上发射的所有流来扁平传输流。Flatmap不是修复程序，传输流也不是bug，它们确实是用来处理Rx中异步响应的工具。

不错啊。因为响应流是根据请求流定义的，所以如果稍后我们有更多的事件发生在请求流上，我们就会像预期的那样在响应流上发生相应的响应事件：

现在我们终于有了响应流，我们可以呈现我们收到的数据了：

Var requestStream=Rx。看得见。只需(#39；https://api.github.com/users'；)；var responseStream=requestStream。FlatMap(function(RequestUrl){return Rx.。看得见。FromPromise(jQuery.。GetJSON(RequestUrl))；})；responseStream。Subscribe(function(Response){//Render`Response`随心所欲地访问DOM})；

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

每次单击刷新按钮时，请求流都应该发出一个新的URL，这样我们就可以获得一个新的响应。我们需要两件事：刷新按钮上的单击事件流(咒语：任何东西都可以是流)，并且我们需要更改请求流以依赖于刷新点击流。令人高兴的是，RxJS附带了一些工具，可以从事件侦听器获得可观察性。

因为刷新单击事件本身不携带任何API URL，所以我们需要将每次单击映射到一个实际的URL。现在，我们将请求流更改为每次使用随机偏移量参数映射到API端点的刷新点击流。

Var requestStream=fresh hClickStream。Map(function(){var随机偏移=数学。地板(数学。Random()*500)；返回'；https://api.github.com/users?since='；+随机偏移；})；

因为我很笨，而且我没有自动测试，所以我刚刚破坏了我们之前构建的一个功能。请求在启动时不再发生，它只在单击刷新时发生。呃。我需要这两种行为：单击刷新或刚刚打开网页时的请求。

Var requestOnRechresStream=fresh hClickStream。Map(function(){var随机偏移=数学。地板(数学。Random()*500)；Return'；https://api.github.com/users?since='；+随机偏移；})；var启动pRequestStream=Rx。看得见。只要(#39；https://api.github.com/users'；)；

但是我们怎样才能把这两件事合而为一呢？嗯，这是Merge()。用图表方言解释，它是这样做的：

Var requestOnRechresStream=fresh hClickStream。Map(function(){var随机偏移=数学。地板(数学。Random()*500)；Return'；https://api.github.com/users?since='；+随机偏移；})；var启动pRequestStream=Rx。看得见。只需(#39；https://api.github.com/users'；)；var requestStream=Rx。看得见。Merge(requestOnRechresStream，StartupRequestStream)；

Var requestStream=fresh hClickStream。Map(function(){var随机偏移=数学。地板(数学。Random()*500)；返回'；https://api.github.com/users?since='；+随机偏移；})。合并(处方。看得见。就(#39；https://api.github.com/users'；)；

Var requestStream=fresh hClickStream。Map(function(){var随机偏移=数学。地板(数学。Random()*500)；返回'；https://api.github.com/users?since='；+随机偏移；})。开始(#39；https://api.github.com/users'；)；

StartWith()函数的作用与您认为的完全相同。无论您的输入流是什么样子，startWith(X)得到的输出流的开头都是x。但是我还不够干，我正在重复API终结点字符串。解决这个问题的一种方法是将startWith()移到接近fresh hClickStream的位置，实质上模拟启动时的刷新单击。

Var requestStream=fresh hClickStream。StartWith(启动单击)。Map(function(){var随机偏移=数学。地板(数学。Random()*500)；返回'；https://api.github.com/users?since='；+随机偏移；})；

好的。如果回到我破坏自动化测试的地方，您应该会看到，最后一种方法的唯一区别是我添加了startWith()。

到目前为止，我们只在ResponseStream的Subscribe()的呈现步骤中触及了Suggest UI元素。现在使用“刷新”按钮，我们遇到了一个问题：一旦您单击“刷新”，当前的3条建议就不会被清除。新的建议只有在响应到达后才会出现，但是为了使UI看起来更美观，我们需要在刷新时清除当前的建议。

不，别这么快，伙计。这很糟糕，因为我们现在有两个影响建议DOM元素的订阅者(另一个是responseStream.scribe())，这听起来并不像是关注点分离。还记得那句反应性的咒语吗？

因此，让将建议建模为流，其中发出的每个值都是包含建议数据的JSON对象。我们将针对这3个建议中的每一个分别执行此操作。这就是这条街是如何。

.