Part 1 初入反应式扩展

什么是反应式程序？它们有什么用？使用反应式扩展（Rx）编程，会如何改变你编写代码的方式？在开始使用 Rx 之前应该做些什么？为什么 Rx 比传统的事件驱动编程更好？ 这些都是我们将在前三章开始讨论的问题。

你将了解什么是反应式系统及反应式程序，以及为什么要关注它们。您将看到创建使用 Rx 的应用程序的真实示例，以及创建自己的 Rx 程序所需要做的事情。你还将了解 Rx 所基于的函数式编程基础，从而更容易理解本书介绍的其他概念。

1 反应式编程

本章内容包括

• 反应式编程思维 (Being reactive)
• 将事件视为流
• 引入反应式扩展 (Rx)

近年来，反应式编程模式越来越受欢迎，它旨在简化事件驱动应用程序的构建和异步代码执行。 反应式编程专注于变化的传播及其影响——简单地说，就是如何对变化做出反应，并创建依赖于变化的数据流。

随着 Facebook 和 Twitter 等应用程序的兴起，大洋彼岸发生的每一个变化（例如状态更新）都会立即被另一端观察到，并在应用程序内部立即产生一连串的反应。因此，需要一个简化的模型来表达这一反应链也就不足为奇了。如今，外部环境的变化（如 GPS 定位、电池和电源管理以及社交网络消息）以及应用程序内部的变化（如网络调用响应、文件读写和计时器）对现代应用程序的驱动力很大。对于所有这些事件，应用程序都会做出相应的反应——例如，更改显示的视图或修改存储的数据。

我们发现，在机器人技术、移动应用程序、医疗保健等多种类型的应用程序中，都需要一个简化的事件反应模型。以类似命令方式对事件做出反应会导致代码繁琐、难以理解且容易出错，程序员面对协调事件和数据更改时，需要手工修改功能相对独立的代码块，而这些代码却更改相同的变量或者数据（可以简单理解为不同函数都要去更改某个公共变量）。而这些变化发生的顺序不可预测，有时甚至同时发生。反应式编程把事件和随时间变化的状态进行了抽象，我们可以给将要发生的事件定义一个执行链，有了这个执行链，我们就不用再去考虑“如何管理这些值之间的依赖关系”了，从根本上把我们解放出来。
反应式扩展（Rx）是一个为.NET程序提供反应式编程模型的库。Rx通过使用声明性操作（采用LINQ风格）在单个事件序列上创建查询，从而使事件处理代码更简单且更具表达性。Rx还提供了称为组合器(combinators)（组合操作）的方法，使您能够连接事件序列，以处理事件发生的模式或它们之间的关联。在撰写本文时，Rx库中有超过600个操作（包括重载）。每个操作都封装了经常出现的事件处理代码，否则您可能需要自己编写。
本书的目的是教你为什么要接受反应式编程的思维方式，以及如何使用 Rx 轻松地构建事件驱动型应用程序，最重要的是，这本书充满了乐趣。本书将逐步向你介绍 Rx 的各个层面，先从构建基本模块开始，该模块支持你创建反应式数据流和事件流。然后展示Rx 提供的丰富查询功能，演示Rx并发模型如何帮你控制代码的异步性，并控制你的反应式处理器(Operator)。但是，您首先需要了解反应式的含义，以及传统命令式编程与处理事件的反应式方法之间的区别。

1.1 反应式编程思维

当应用程序发生变化时，您的代码需要对其做出反应；这就是反应性的含义。变化有多种形式。最简单的一种就是我们在日常编程中习以为常的可变值的变化。变量持有的值可以在特定时间通过特定操作进行更改。例如，在 C# 中可以这样写：

int a = 2;
int b = 3;
int c = a + b;
Console.WriteLine("before: the value of c is {0}",c);
a=7;
b=2;
Console.WriteLine("after: the value of c is {0}",c);

输入如下

before: the value of c is 5
after: the value of c is 5

小程序中，有两处打印输出c，c 都显示了相同值。在我们的命令式编程模型中，c 的值是 5，除非你显式地修它，否则它将保持 5。

有时，您希望在 a 或 b 发生变化的时候更新 c。为此，反应式编程引入了一种变量类型——时变变量(time varying)：这种变量的值并不固定，而是会随着时间的变化而变化。
再来看看我们的小程序；如果它在反应式编程模型中运行时，输出应该是：

before: the value of c is 5
after: the value of c is 9

c 的值 "神奇地 "发生了变化。这是因为它的依赖关系发生了变化。如图 1.1 所示，这个过程的工作原理就像一台机器从两条平行的传送带上输送物料，并从两边的输入端生产出一个物品。

图 1.1 函数 c = a + b 的反应式表示。当 a 为 7，b 为 2 时，c 自动变为 9。当 b 变为 1 时，c 变为 8，因为 a 的值仍然是 7

你可能会觉得奇怪，但你可能已经使用反应式应用程序多年了。正是这种反应性概念让你最喜欢的电子表格应用程序使用起来如此简单有趣。在电子表格单元格中创建这种等式时，每次改变等式输入单元格中的值，最终单元格中的结果都会自动改变。

程序中存在的反应式

在实际应用中，您可以在许多情况下发现可能的时间变量，例如 GPS 位置、温度、鼠标坐标，甚至文本框内容。所有这些变量都有一个随时间变化的值，应用程序会对其做出反应，因此它们都是时间变量。值得一提的是，时间本身也是时间变量；它的值一直在变化。在 C# 等命令式编程模型中，您可以使用事件来创建对变化做出反应的机制，但这会导致代码难以维护，因为事件分散在不同的代码片段中。
想象一下，有一款移动应用程序，她能帮助用户找到周边商店的折扣和特价商品。我们称之为 Shoppy。图 1.2 描述了 Shoppy 的架构。

如图 1.3 所示，您希望 Shoppy 提供的一个强大功能：当用户越接近商店的区域（该区域是以商店为中心辐射出来的一个圆），地图上商店图标的尺寸就越大。此外，您还希望系统能在更新时向应用程序推送新的交易。

在这种情况下，可以说 store.Location(商店位置)、myLocation(我的位置) 和 iconSize(图标大小)，这三个变量都是时间变量。对于每个商店来说，图标大小公式为：

//注意如下代码中的变量不是普通变量,要想象成时变变量
distance = store.Location – myLocation; //计算我与某商店的距离
iconSize = (MINIMAL_RADIUS / distance)*MinIconSize //使用距离计算该商店图标大小

由于使用了时变变量，每次 myLocation 变量发生变化时，都会触发距离变量distance的变化。应用程序最终会根据商店的距离，使商店图标变大或变小。 请注意，为了简单起见，我没有对最小图标尺寸进行界定，因此距离为 0 或接近 0时图标大小可能计算出来非常大。
这只是一个简单的例子，但正如你所看到的，使用反应式编程模型的强项在于它能够组合、连接、分割和拆分每个时变变量推送的数据流。 这是因为反应式编程可以让你专注于要实现的目标，而不是实现目标的技术细节。这使得代码简单易读，并消除大多数模板代码（如变更跟踪或状态管理），模板代码会分散你在编写代码时，对原本意图或逻辑的注意力。当代码简而精炼时，错误就会减少，也更容易掌控。
现在可以停止理论介绍，让我们看看如何借助Rx ，在 .NET 中实现反应式编程。

1.2 介绍反应式扩展(Reactive Extensions)

既然我们已经介绍了反应式编程，那么是时候认识一下我们的明星： Reactive Extensions，通常简称为 Rx。 微软开发 Reactive Extensions 库的目的，是为了方便处理事件流和数据流。从某种程度上讲，时变值本身就是一个事件流；每个值的变化都是一种事件，你可以订阅它，并更新依赖于它的值。
Rx 将事件流抽象为Observable(可观察序列)，从而方便了事件流的处理，这也是 Rx 表达时变值的方式。Observable(可观察序列)意味着用户可以观察到序列所携带的值，而序列则意味着所携带的值是有顺序的。Rx 由 Erik Meijer 和 Brian Beckman 设计，灵感来自函数式编程风格。在 Rx 中，某某流由Observable(可观察序列)表示，Observable(可观察序列)可以根据 .NET 事件、Task(任务)或Collection(集合)创建，也可以自己从其他来源创建。使用 Rx，您可以使用 LINQ 处理器查询Observable(可观察序列)，并使用 Scheduler(调度器)控制并发； 这就是为什么在 Rx.NET 中，经常将 Rx 定义为 Rx =Observable(可观察序列) + LINQ + Schedulers。 Rx.NET 的层如图 1.4 所示。

在本书中，你将探索 Rx 分层中每个组成部分，以及它们之间的相互作用，但首先让我们了解一下 Rx 起源的简史。

学习笔记：这一段非常非常重要，作者在这里说了一个很重要的概念——事件流，英文是 streams of events。在Rx中，事件流被理解成一种特殊类型，英文写作Observable。
Observable这个词用中文翻译不好表达，她本意是个形容词，表示可以被观察到地，但在这里是一个名词，目前称之为可观察序列，可以理解为：一个可被监听或可被观察或可被订阅的集合体对象，其中重要的概念就是两个：1可被监听，2这是一个集合体，这个词尾的able强调是可以被怎么怎么样，重点强调可以，词头的observer，是动词观察的意思。
作者在后面的章节中会展开介绍Observable，这里我给出自己对她的通俗理解：

• Observable是一个集合，类似数组或者集合对象等等，当然她也可以只有一个元素。
• Observable中存储的每一个元素，本质上应该理解为事件，或者说抽象的理解成事件。
• Observable她可以根据事件创建，根据Task创建，根据集合创建，甚至能根据定时器创建。当构建好一个Observable后，里面的元素可以动态的加入进去。而加入进去的过程可以是异步的，比如来自消息，消息发生后往里面增加一个元素，或者来自io任务结束，io结束后把结果整理成一个元素放进去，来自Task执行结束，或者来自定时器，比如每隔5秒产生一个元素。
• Observable就像有生命一样，只要有元素进入到Observable里面，她就会向订阅者进行告知，告诉对方我收到新元素了，你来处理吧。
  
  举个例子，假设现在你能得到一个Observable<int>的对象，你就应该知道，这里面肯定是存储了一堆int数，也有可能是1个啊。但是我们不能把它理解成List，而应该理解成是一个有着先后顺序的整数数据事件流。这词很绕口，大白话说，就是这里面是一个不定时会产生信号或者事件的队列，里面存储是将来要发生的事情，每个事情用一个int表达。你要问Observable<int>此时此刻里面有数据吗，对不起！这个无法回答，但是你可以放一个监视器在他身上，只要里面有数据了，监视器就会得到，然后你可以在监视器里处理数据。并且你处理后可以将结果扔回Observable里面,等待后面的人处理。
  这个过程让我想到了，卓别林的电影摩登时代，Observable就是那条长长的流水线，指不定什么时候来一个活就需要你干，干完了你这一步的工序，后面会有人继续其他工序，直至整个流水线完结产出成品。
  

图中的整条流水线就像是Observable，卓别林和他的两个工友就是观察者或者叫监听者，英文是Observer。

1.2.1 Rx历史

我认为，想要完全掌控一件事物（尤其是技术），应知其历史和细节。让我们从 Rx 的徽标开始，如图 1.5 所示，徽标上有一条电鳗；这条鳗鱼是微软 Live Labs 的 Volta 项目徽标。

Volta 项目是一个实验性的开发人员工具集，用于在云正式定义之前为云创建多层应用程序。使用 Volta，您可以指定应用程序的哪一部分需要在云中运行（服务器），哪一部分需要在客户端运行（桌面、JavaScript 或 Silverlight），Volta 编译器会为您完成这些繁重的工作。很快，我们发现在将服务器上发生的事件传输到客户端方面存在差距。由于 .NET 事件不是一等公民，它们无法序列化并推送到客户端，因此，Observable(可观察序列)和观察者这对组合应运而生（虽然当时还没有这个名称）。

Rx 并不是 Volta 项目产生的唯一技术。它还发明了 JavaScript 编译器的中间语言（IL），这也是微软 TypeScript 的起源。开发 Volta 的团队也是 Rx 诞生的团队。 自 2010 年发布以来，Rx 已成为一个成功的故事，被许多公司采用。.NET之外的其他社区也看到了它的成功，并很快将其移植到其他语言和技术中。 例如，Netflix 在其服务层中广泛使用 Rx，并负责 RxJava 移植。4 微软内部也使用 Rx 来运行 Cortana——每个 Windows Phone 设备中都有的智能个人助理；当你创建一个事件时，后台就会创建一个 Observable(可观察序列)。

在撰写本文时，Rx 支持 10 多种语言，包括 JavaScript、C++、Python 和 Swift。Reactive Extensions 现在是一系列开源项目的集合。您可以在 http://reactivex.io/ 上找到有关这些项目的信息、文档和新闻。.NET的Reactive Extensions托管在GitHub repo下：https://github.com/Reactive-Extensions/Rx.NET。
现在，我们已经介绍了一些历史，并讲述了它的生存之道，让我们开始探索 Rx 的内部结构吧。

1.2.2 客户端和服务器上的 Rx

Rx 非常适合事件驱动型应用程序。之所以这样，是因为事件（如前所述）是创建时变值的必要方式。 从历史上看，事件驱动编程主要出现在客户端技术中，因为用户交互是通过事件来实现的。例如，你可能使用过 OnMouseMove 或 OnKeyPressed 事件。故此许多客户端应用程序都使用 Rx。 此外，一些客户端框架也基于 Rx，例如 ReactiveUI (http://reactiveui.net)。
但我要告诉大家的是，Rx 并不只是客户端技术。而且，在服务器端许多应用场景的代码中，Rx 完全胜任。此外，如我之前所述，Rx 还可用于大型应用，如 Microsoft Cortana、Netflix 和使用 Microsoft StreamInsight 的复杂事件处理 (CEP)。不管在服务层还是客户端层，Rx 在处理应用程序接收消息方面，是非常优秀的库。

1.2.3 Observables

Observable 用于在 Rx 中实现时变值（我们将其定义为可观察序列）。它代表推送模型，其中的新数据被推送给（或通知）observer(观察者)。
Observable 被定义为事件（或通知）源，如果你愿意，也可以定义为数据流的发布者。推送模式意味着观察者不必从数据源获取数据，也不必总是检查是否有新的数据尚未被获取（拉动模式），而是在数据可用时将其发送给观察者。
Observable实现了 IObservable 接口，该接口自 .NET Framework 4.0 版以来一直位于 System 命名空间中。

Listing 1.1 IObservable interface

public interface IObservable<T>
{
    //将Observer Subscribe到 Observable中
    //将(观察者) (订阅)    到 (可观察序列)中
    IDisposable Subscribe(IObserver<T> observer);   
}

IObservable 接口只有一个方法 Subscribe，允许观察者订阅通知。 Subscribe 方法返回一个 IDisposable 对象，该对象代表订阅，并允许观察者随时调用 Dispose 方法取消订阅。Observable(可观察序列) 保存着已订阅观察者的集合，并在有值得通知的事情时通知它们。这是通过 IObserver 接口实现的，如图所示，该接口自 .NET Framework 4.0 版以来一直位于 System 命名空间中。

Listing 1.2 IObserver interface

public interface IObserver<T>
{
    void OnNext(T value); //IObservable中出现了一个新元素时,会调用Observer.OnNext    
    void OnError(Exception error); //IObservable告知Observer发生异常了
    void OnCompleted(); //IObservable告知Observer,元素全部推送完毕了,以后不会再有新元素过来了    
}

使用 IObservable 和 IObserver 的基本流程如图 1.6 所示。Observable(可观察序列)并不总是完整的；它们可以是潜在的无限制序列元素（如无限集合）的提供者。Observable(可观察序列)也可能是 “安静的”，这意味着它从未推送过任何元素，也永远不会推送。Observable(可观察序列)也可能失败；失败可能发生在观察对象已经推送元素之后，也可能在没有推送任何元素的情况下发生。
下面的表达式(其中 * 表示零次或多次，? 表示零次或一次，| 是 OR 运算符)正式表述了这种可观察序列：

$$OnNext(t)\ast (OnCompleted()|OnError(e))?$$

失败时，将使用 OnError 方法通知观察者，并将异常对象传递给观察者进行检查和处理（见图 1.7）。 出错后（以及完成后），将不再向观察者推送消息。当 Observable(可观察序列)没有提供错误处理器(Operator)时，Rx 使用的默认策略是升级异常并导致崩溃。你将在第 10 章中了解优雅处理错误的方法。

观察者设计模式(The Observer design pattern)
在某些编程语言中，事件有时被作为一等公民提供，这意味着你可以使用语言提供的关键字、类型定义、注册事件，甚至可以将事件作为参数传递给函数。对于不支持将事件作为一等公民的语言，观察者设计模式是一种非常有用的设计模式，它可以为应用程序添加类似事件的支持。此外，.NET 的事件实现也是基于这种模式。
观察者设计模式是由四人帮（Gang of Four，GoF）在《设计模式》（Design Patterns）一书中提出的： Elements of Reusable Object-Oriented Software》（Addison-Wesley Professional，1994 年）中提出的。该模式定义了两个组件：主体和观察者（不要与 Rx 的 IObserver(观察者)混淆）。Observer是对事件感兴趣的参与者，它将自己订阅给引发事件的 subject。这是统一建模语言（UML）类图中的样子：

观察者设计模式很有用，但也有几个问题。Observer只有一个接受事件的方法。如果要附加到多个 subject 或多个事件，就需要实现更多的更新方法。另一个问题是，该模式没有给出处理错误的最佳方法，如果有的话，也要由开发人员来寻找通知错误的方法。最后还有个小问题——如何知道subject已经结束，subject结束后是不会再有通知发送出来的，这对于资源管理可是至关重要的。Rx IObservable 和 IObserver 基于观察序列设计模式，但对其进行了扩展，以解决这些缺陷。

1.2.4 处理器(Operator)

Reactive Extensions 还带来了丰富的处理器(Operator)。在 Rx 中，处理器(Operator)对应操作方法，它以声明的方式来描述事件处理，并且它也是特定领域语言（DSL）的一部分。通过 Rx 处理器(Operator) ，您可以使用Observable(可观察序列)和Observer(观察者)创建查询管道，并对其进行查询、转换、投影和其他 LINQ 中已知的事件处理。 Rx 库还包括时间基（time-based）处理器，和 Rx 专用处理器，用于查询、同步化、错误处理等。
例如，您可以订阅一个字符串类型的Observable(可观察序列)，仅处理序列中 A 开头的字符串，并将字符串转换为大写字母：

    IObservable<string> strings= ...   //字符串类型的IObservable将其中的每一个字符串推送给Observer             
    IDisposable subscription = //订阅(subscription)对象，用后可以注销。                      
    strings.Where(str => str.StartsWith("A"))  //只允许向Observer传递以 A 开头的字符串
        .Select(str => str.ToUpper())  //字符串转换为大写,并提供给后续。         
        .Subscribe(...); //Observer订阅接收过滤和转换后的字符串。                 
    //Rest of the code
    :
    subscription.Dispose(); //当您不想再接收字符串时，注销订阅(subscription )。

注意 如果不理解代码中的语法或关键字，也不要害怕。接下来的章节中我会一一讲解。

在这个简单的示例中，你可以看到 Rx 声明式的处理器(Operator)，表达非常直白，因此代码读起来就像一个故事。这个示例中，重点只是介绍一下查询处理器(Operator)，所以我没有展示如何创建 Observable(可观察序列)。 你可以通过多种方式创建Observable(可观察序列)：事件、枚举、异步类型等。具体将在第 4 章和第 5 章中讨论。现在，我们可以假设Observable是在幕后创建好的。
处理器(Operator)和组合器（组合多个Observable的处理器(Operator)）可以帮助你处理多Observable触发事件的复杂场景。想实现 Shoppy 示例中自动调整商店图标的大小功能，可以编写以下 Rx 代码：

IObservable<Store>  stores = ... //提供商店信息的Observable           
IObservable<Location>  myLocation = ...//提供当前位置的Observable        
IObservable<StoreIconSize> iconSize = 
               from store in stores //处理每一个商店信息,stores中存储的每个元素都是Store类型           
               from currentLocation in myLocation //与商店(stores) Observable类似。每次位置发生变化时，都会获取所有商店和当前位置的数据对。
               let distance = store.Location.DistanceFrom(currentLocation) //根据商店和位置再创建两个新变量。
               let size = (MINIMAL_RADIUS / dist) * MIN_ICON_SIZE //分别计算商店的距离(distance)和商店图标的大小(size)。
               select new StoreIconSize { Store=store , Size=size };    //将计算后的大小和商店包装成一个新的元素StoreIconSize
//括号中的 lambda 表达式充当Observer的 OnNext 实现，每次商店图标有新尺寸时都会被调用。               
iconSize.Subscribe( iconInfo => iconInfo.Store.Icon = iconInfo.Size);  

即使不了解反应式扩展的所有细节，你也可以看到，实现 Shoppy 程序所需功能的代码很少，而且容易阅读。处理各种数据流的代码都是由 Rx 完成的。与以往编写事件处理程序相比，减轻了很多负担，以往的程序写法是根据事件分割成很多独立的处理函数，代码非常碎而且分散。

1.2.5 Rx 处理器(Operator)的嵌套

大多数 Rx 处理器(Operator)的格式如下：

IObservable<T> OperatorName(参数)

请注意，这里返回类型是 Observable(可观察序列)。如此，就实现了 Rx 处理器(Operator)的嵌套；你可以将处理器(Operator)加入到 Observable管道(Observable pipeline) 中，每一次加入都会产生一个新的 Observable，并保持与源头对象一致的通知行为，即内层Observable产生元素后，外层Observable也会层层感知。
另一个重要的启示是，从Observer（观察者）的角度来看，无论是否添加了处理器(Operator)， Observable 仍然是 Observable，如图 1.8 所示。
因为你不仅可以在创建Observable对象时向管道中添加处理器(Operator)，还可以Observable已经被订阅后，进行处理器(Operator)添加，所以即使您无法访问创建Observable对象的代码，也可以控制Observable对象。

1.2.6 弹珠图 （Marble diagrams）

一图胜千言。这就是为什么在讲解反应式编程，尤其是 Rx 时，展示 Observable(可观察序列)的执行流水非常重要。在本书中，我使用**弹珠图（Marble diagrams）**来帮助你理解操作和他们之间的关系。
弹珠图使用水平轴来表示 Observable(可观察序列)。Observable(可观察序列)上承载的每个通知都用一个符号标出，通常是一个圆圈（有时也会用其他符号），以区分不同的值。
如图 1.9 所示，通知的值写在符号内或符号上方的注释中。

在弹珠图中，时间从左到右，符号之间的距离表示两个事件之间经过的时间。距离越长，时间就过得越久，但这只是相对而言。我们无法知道时间的单位是秒、小时还是其他计量单位。如果这些信息很重要，就会以注释的形式写出来。
要表示观察对象已完成，可以使用 | 符号。图 1.10 显示了示例。

要显示一个操作符（或多个操作符）在可观察序列上的输出，可以使用箭头来表示源事件和结果之间的关系。请记住，每个操作符（至少绝大多数操作符）都会返回自己的观察序列，因此在图中，我把作为流水线一部分的操作符写在左侧，而代表从它返回的观察序列的线则写在右侧。图 1.11 显示的是上例中观察字符串序列的大理石图，它只显示以 A 开头的字符串，并将其转换为大写。

本书使用弹珠图来显示处理器的执行效果，以及嵌套处理器创建Observable管道的示例。说到这里，你可能想知道可Observable(可观察序列)与Nonobservable(不可观察序列)的关系。 接下来就是答案。

1.2.7 Pull model vs. push model

Nonobservable(不可观察序列)就是我们通常所说的迭代器enumerables（或集合collections），它实现了 IEnumerable 接口，并返回一个实现了 IEnumerator 接口的迭代器。使用迭代器时，通常是通过循环从集合中提取值。 Rx Observable的行为则不同：不是拉取(Pull)，而是将值推送(Push)给Observer(观察者)。表 1.1 和 1.2 显示了拉和推模型之间的对应关系。二者之间的这种关系称为对偶原理(duality principle)。

表 1.1 IEnumerator 和 IObserver 如何相互对应

IEnumerator	IObserver
MoveNext—when false	OnCompleted
MoveNext—when exception	OnError(Exception exception)
Current	OnNext(T)

表 1.2 IEnumerable 和 IObservable 如何相互对应

IEnumerable	IObservable
IEnumerator GetEnumerator(void)	IDisposable Subscribe(IObserver)

这里的二元性有一个例外，因为 GetEnumerator 参数的孪生参数（是 void类型）本应转换为 Subscribe 方法的返回类型（并保持 void），但却使用了 IDisposable。

Observable和Observer填补了.NET在处理异步操作的空白，异步操作需要在推送模型中返回一串值（即推送序列中的每个项目）。与Task异步提供单值 ，和 IEnumerable 同步拉取多值相比，Observable(可观察序列) 则是异步发送出来一个序列。表 1.3 对此进行了总结。

表 1.3 推送模式(Push model) 与 拉取模式(pull model) 的数据类型

	Single value	Multiple values
拉取/同步/非反应式	T	IEnumerable
推送/异步/反应式	Task	IObservable

由于Observable和Enumerable之间存在反向对应关系（二元性），因此您可以将数值序列从一种表示方法转换为另一种表示方法。比如固定的集合，如 List，可以转化为一个Observable(可观察序列)，将其所有值推送给Observer(观察者)。更令人惊讶的是，Observable(可观察序列) 可以转化为基于拉取的集合。在后面的章节中，我们将深入探讨如何，以及何时进行这些转换。现在最重要的是要明白，因为你可以将一种模型转化为另一种模型，所以所有拉取模型的操作，可以直接作用于推送模型。因此，当你遇到问题时，可以用最简单的模型来解决，然后根据需要转换结果。
我在这里要说的最后一点是，因为你可以把单个值看成是仅有一个项目的集合，所以你也可以用同样的逻辑，把异步单个元素的Task，看成是仅有一个元素的Observable，反之亦然。请牢记这一点，因为这是理解 "万物皆可观察 "的重要一点。

学习笔记：笔者在这小节其实想说的事情还是很简单的，但是他用的语言比较严谨。看着非常绕。
我个人理解就是，在使用IEnumerable里的数据时，一般都需要用foreach语句去主动提取出来。这个过程他叫做拉取，英文叫Pull。并且这个操作是同步的。更诗意的说法：值就在那里，不增不减，你若去拉，他就会出来。
而Observable就不一样，虽然他也像是一个集合类，但是里面的值拉取不出来，根本原因就是这玩意的值是后期异步产生的，你也不知道啥时候会出来值，所以只能等着Observable把值推送出来，英文叫做Push。
然后又说其实万物都可以转换成Observable，因为值提前放在里面，还是后期异步产生，对Observable来说都可以，反正Observable早晚会推送出来自己序列中的所有元素给Observer。
接着又解释了一下单值是也是集合，只是一个数量为1的集合罢了，所以把Task看成了单元素Observable。
至于对偶性这个问题，我个人只能理解成，作者想用一种数学方式证明IEnumerable和Observable存在一种对等关系，并可以相互转换。或者说可以套用linq那一套东西，毕竟linq本身是作用在IEnumerable上的，只有证明了Observable和IEnumerable存在某种对等关系，才能把linq套在Observable上使用。

1.3 使用反应式系统和反应式宣言

到目前为止，我们已经讨论了 Rx 如何为应用程序增加反应能力。许多应用程序并不是独立的，而是整个系统的一部分，该系统由更多的应用程序（桌面、移动、网络）、服务器、数据库、队列、服务总线和其他组件组成，您需要将它们连接起来才能创建一个工作的有机体。反应式编程模型（以及作为该模型实现的 Rx）简化了应用程序处理变化传播和事件消耗的方式，从而使应用程序成为反应式的。但如何才能使整个系统具有反应性呢？
作为一个系统，反应性的定义:即时响应性（Responsive）、回弹性（Resilient）、弹性（Elastic）以及消息驱动（Message Driven）。反应式宣言(Reactive Manifesto)（www.reactivemanifesto.org）定义了反应式系统的这四个特征，这是软件社区为定义构建最优架构风格反应式系统，而做出的共同努力。您可以通过签署宣言和传播信息，加入到提高人们对反应式系统认识的行列中来。
重要的是要明白，《反应式宣言》并没有发明任何新东西；早在宣言发布之前，反应式应用程序就已经存在了。例如，电话系统已经存在了几十年。这种分布式系统需要对动态负载量（呼叫）做出反应，从故障中恢复，并对呼叫者和被呼叫者保持全天候的可用性和响应性，而所有这些都是通过一个操作员向另一个操作员传递信号（信息）来实现的。
宣言的目的是为反应式系统这一概念构建一版图，收集搭建反应式系统的最佳实践。让我们深入探讨这些概念。

1.3.1 响应 Responsiveness

当您在心仪的浏览器上输入 URL 时，您会希望在很短的时间内加载您要浏览的页面。如果加载时间超过很多毫秒，你会有一种不祥的预感（甚至会生气）。可能会决定离开该网站，到其他网站浏览。如果你是网站所有者，你就失去了一个客户，因为你的网站没有响应能力。
系统的响应速度，取决于系统对收到的请求后，做出响应的时间。显然，响应时间越短，说明系统的响应速度越快。系统的响应可以是积极的结果，例如您试图加载的页面、试图从网络服务中获取的数据或您想在金融客户端应用程序中看到的图表。响应也可以是负面的，例如一条错误信息，说明你输入的某个值无效。
无论是哪种情况，如果系统做出响应所需的时间是合理的，那么就可以说应用程序是有响应的。但合理的时间很难定义，因为它取决于具体情况和所测量的系统。对于有一个按钮的客户端应用程序来说，可以假定应用程序的响应是点击按钮需要多少毫秒。对于需要进行大量计算的网络服务来说，一到两秒也是合理的。在设计应用程序时，您需要分析现有的操作，并确定操作完成和响应所需的时间范围。反应灵敏是反应式系统努力实现的目标。

1.3.2 弹性 Resiliency

每隔一段时间，您的系统就可能面临故障。如网络断开、硬盘驱动器故障、电力中断或内部组件出现异常情况等。弹性系统就是在发生故障时仍能保持响应的系统。换句话说，在编写应用程序时，要以不妨碍用户获得响应的方式处理故障。
为应用程序添加弹性的方式因应用程序而异。比如一个程序去捕获异常，并将程序状态退回到常规状态。而另一个程序应添加更多的服务器，如此，万一其中一个服务器崩溃，另一个服务器就会接替，并处理请求。若要提高系统弹性，应遵循一良好原则——即避免单点故障。为此可以将应用程序的每个部分与其他部分进行分离；您可以将各部分区分成不同的 AppDomains、不同的进程、不同的容器或不同的机器。通过分离各部分，可以使系统整体不可用的风险降低。

1.3.3 伸缩性 Elasticity

您编写的应用程序将被许多用户使——最好是大量用户。每个用户都会向系统发出请求，这可能会导致系统需要处理负载很高。系统中的每个组件都负载极限，当负载超过极限时，请求将开始失败，组件本身也可能崩溃。当系统遭受分布式拒绝服务 (DDoS) 攻击时，也会导致负载增加。
为了克服过载的原因，系统需要具有伸缩能力：当负载增加时，系统需要扩增实例；当负载减少时，系统需要缩减实例。自从云技术进入我们的生活后，这种自动行为变得更加明显。在云上运行时，你会产生资源无限的错觉；只需几个简单的配置，你就可以根据自己定义的阈值来设置应用程序的扩展或缩减。您只需记住，运行额外的服务器是需要成本的。

1.3.4 消息驱动 Message driven

如果说及时响应(Responsiveness)是你的终极目标，弹性(Resiliency)是确保及时响应的方法，而伸缩性(Elasticity)则是保持弹性(resilient)的一种方法。那么在反应式系统拼图中，还缺少一个部分，那就是系统各部分之间采用哪种通信方式，以实现我们所探讨的反应式。
其实异步消息传递，是最符合我们需求的通信方式，因为它允许我们控制每个组件的负载水平，而不限制生产者（一般策略是使用队列或服务总线等中间渠道）。它允许将信息路由到正确的目的地，并在组件崩溃时重新发送失败的信息。它还增加了系统内部组件的透明度，因为用户不需要知道系统的内部结构，只需要知道它能处理的消息类型。消息驱动使得所有其他反应式概念成为可能。图 1.12 显示了使用消息队列的消息驱动方法，该方法展示了如何帮助系统提高消息处理速度，并实现弹性和伸缩性。

图中，整个系统借助消息队列,并以消息驱动方式进行通信。客户端发送消息，服务器随后检索消息。这种异步通信模型,给整个系统在处理速度和故障处理方面，提供了很强的控制力。目前有许多消息队列的实现方法，它们具有不同的功能集。有些允许信息的持久保存，从而提供了持久性；有些还提供了 "事务性 "交付模式，即锁定信息，直到消费者发出处理成功的信号。无论选择哪种消息队列（或消息驱动平台），都需要去获取已发送的消息并处理它们。这就是 Rx 的功能所在。

1.3.5 Where is Rx?

反应式扩展库在构建反应式系统方面，扮演着重要的角色，它与消息驱动的概念有关。Rx的功能并不是在应用程序或服务器之间移动消息，而是在消息到达时，负责处理消息并在应用程序内部沿着执行链传递消息的机制。需要说明的是，即使你不是在开发一个包含许多组件的完整系统，也可以使用 Rx。即使是单个应用程序，也能发现 Rx 在事件处理上，和程序响应式处理消息方面都非常有用。图 1.13 揭示了所有《反应式宣言》概念与 Rx 之间的关系。

要构建一个完整的反应式系统，图中的所有概念都必须存在。在不同的系统中，每个概念的实现方式都不尽相同。Rx 是一种方法，一种可以更方便地接收消息的方法，因此它是消息驱动模块的一部分，所以显示在消息驱动模块中。Rx 的引入，恰恰是她提供了异步处理和基于事件编程的方法，如本例中的消息，所以我必须解释一下异步含义以及它的重要性。

1.4 了解异步性

异步消息传递是反应式系统的关键特征。但异步究竟是什么，为什么它对反应式程序如此重要？我们的生活由许多异步任务组成。其实你可能没有意识到，但如果日常生活不是异步的话，那就烦人了。要了解什么是异步性，首先需要了解非异步执行或同步执行。

**同步(Synchronous)**的定义：发生、存在或产生的过程都在同一瞬间。

同步执行是指你必须等待一项任务完成后，才能继续执行下一项任务。现实中，同步执行的案例就发生在快餐店：你走近柜台面对店员，然后开始想点什么菜，此时店员在一旁等着你，等你点好菜后，需要等厨师准备餐食。最终店员会等你付款成功后，把菜或者餐食给你。到此，你能继续下一任务，那就是去餐桌用餐。这种顺序如图 1.14 所示。

这种顺序给人的感觉就是浪费时间（更确切地说，是浪费资源），所以想象一下，当你为应用程序做同样的事情时，应用程序会有什么感觉。下一节将对此进行演示。

1.4.1 关键在于资源利用
试想一下，如果你在做其他事情之前，必须等待每一项操作完成，你的生活将会是什么样子。设想一下资源因等待和使用的消耗。同样的问题存在于计算机科学中：

writeResult = LongDiskWrite(); //时间比较长的磁盘操作函数
response = LongWebRequest();//时间比较长的Web读取函数
entities = LongDatabaseQuery();//时间比较长的数据库查询函数

在这个同步代码片段中，LongDatabaseQuery 在 LongWebRequest 和 LongDiskWrite 完成之前不会开始执行。在执行每个方法的过程中，调用线程都会被阻塞，资源被占用,实际上是被浪费，占用时无法服务其他请求或处理其他事件。如果这种情况发生在用户界面线程上，程序在执行结束前会一直被冻结。如果这种情况发生在服务器应用程序上，那么在某些时候，您可能会用完空闲的线程，请求也会开始被拒绝。在这两种情况下，程序都会停止响应。
运行上述代码片段所需的总时间如下：

$$total\_time=LongDiskWrit{e}_{time}+LongWebReques{t}_{time}+LongDatabaseQuer{y}_{time}$$
总完成时间等于各部分完成时间的总和。如果能在不等待前一个操作完成的情况下,启动下一个操作，就能更好地利用资源。这就是异步执行的目的。
异步执行是指操作已启动，但其执行是在后台进行的，调用者不会被阻塞。相反，操作完成后会通知调用者。在这段时间内，调用者可以继续做有用的工作。
在点餐示例中，异步方法类似于坐在餐桌上，并有服务员在一旁服务。首先，你坐在桌边，服务员来把菜单递给你，然后离开。在你决定点什么菜的时候，服务员仍然可以为其他顾客服务。当您决定好要吃什么菜后，服务员会回来帮您点菜。在准备食物的过程中，你可以自由地聊天、使用手机或欣赏风景。您不会被阻挡（服务员也不会）。食物准备好后，服务员会把食物端到您的餐桌上，然后继续为其他顾客服务，直到结账时他才会来。
这种模式是异步的：任务并发执行，执行时间与请求时间不同。这样，资源（如服务员）就可以自由处理更多请求。

异步执行发生在哪里？
在计算机程序中，我们可以区分两种类型的异步操作： 基于 CPU 的操作和基于 I/O 的操作。在基于 CPU 的操作中，异步代码在另一个线程上运行，当另一个线程上的执行结束后，返回结果。在基于 I/O 的操作中，操作是在硬盘或网络等 I/O 设备上进行的。在网络上，向另一台机器发出请求（通过使用 TCP、UDP 或其他网络协议），当你机器上的操作系统通过中断从网络硬件获得结果返回的信号时，操作就完成了。 在这两种情况下，调用线程都可以自由执行其他任务，处理其他请求和事件。

异步运行代码的方法不止一种，这取决于所使用的语言。附录 A 显示了在 C# 中可以实现异步运行的方法，并深入介绍了每种方法的位和字节。现在，让我们来看一个使用 .NET 的 Task 类完成异步工作的示例：
上述代码段的异步版本如下所示：

taskA = LongDiskWriteAsync();
taskB = LongWebRequestAsync();
taskC = LongDatabaseQueryAsync();

Task.WaitAll(taskA, taskB, taskC);  

在这个版本中，每个方法都返回 Task。该类表示在后台执行的操作。调用每个方法时，调用线程不会被阻塞，方法会立即返回。然后，在上一个方法仍在执行时，下一个方法会被调用。当所有方法都被调用后，你可以使用 Task.WaitAll 方法等待它们的完成，该方法会获取一个任务集合并阻塞，直到所有任务都完成。另一种写法如下：

taskA = LongDiskWriteAsync();
taskB = LongWebRequestAsync();
taskC = LongDatabaseQueryAsync();

taskA.Wait();
taskB.Wait(); 
taskC.Wait();

这种方法得到的结果是一样的，即等待每个任务完成（当它们仍在后台运行时）。如果在调用 Wait 方法时任务已经完成，它将立即返回。
运行异步版本代码片段所需的总时间如下：
$$total\_time=MAX(LongDiskWrit{e}_{time},LongWebReques{t}_{time},LongDatabaseQuer{y}_{time})$$
由于所有方法都在并发（甚至可能是并行）运行，因此运行代码所需的时间将是最长操作的时间。

1.4.2 异步性与Rx

异步执行并不仅限于使用 Task 来处理。在附录 A 中，您将了解 .NET Framework 中用于提供异步执行的其他模式。
回过头来看 IObservable，它是 Rx 对时变变量的表示，您可以用它来表示任何异步模式，因此当异步执行完成时（成功或出错），执行链将运行，并对依赖关系进行评估。Rx 提供了将各种异步执行类型（如 Task）转换为 IObservable 的方法。
例如，在 Shoppy app中，您不仅想在位置发生变化时获得新折扣，还想在连接状态变为在线时获得新折扣。例如，假设您的手机在短时间内失去信号，然后重新连接。 Shoppy 网络服务的调用是异步进行的，当调用完成后，您希望更新视图以显示新商品：

IObservable<Connectivity> myConnectivity=...  
IObservable<IEnumerable<Discount>> newDiscounts = 
    from connectivity in myConnectivity
    where connectivity == Connectivity.Online
    from discounts in GetDiscounts() //GetDiscounts 返回一个Task,该Task将被转换成Observable
    select discounts;

newDiscounts.Subscribe(discounts => RefreshView(discounts)); //RefreshView显示折扣
private Task<IEnumerable<Discount>> GetDiscounts()
{
    //向服务器发送请求，并接收折扣集合
}

在这个示例中，您要对myConnectivity中的连接变化做出响应。每次连接发生变化时，都会检查您是否在线。如果在线，则异步调用GetDiscounts方法。方法执行完毕后，select语句返回newDiscounts。newDiscounts是一个Observable，最终newDiscounts会推送结果给他的Observer.本例中的Observer就是RefreshView这个函数

1.5 理解事件与流(Stream)

在软件系统中，事件是一种信息类型，用来表示某些事情已经发生。事件可能代表业务事件，例如在图形用户界面应用程序中，按下的每个键或移动的每次鼠标都可能导致事件发生。 事件也可以代表业务事件，例如金融系统中完成的一笔资金交易。
事件由事件源引发，并由事件处理程序处理。正如你所看到的，事件是表示时间变化值的一种方法。而在 Rx 中，事件源可以用可Observable(可观察序列)来表示，事件处理器可以用Observer(观察者)来表示。但那些程序中使用的简单数据呢，例如数据库中的数据或从网络服务器获取的数据。它在反应式世界中占有一席之地吗？

1.5.1 一切皆是流(stream)

您编写的应用程序最终将处理某种数据，如图 1.15 所示。数据有两种类型：静态数据和动态数据。静态数据以数字格式存储，通常从持久存储（如数据库或文件）中读取。数据在网络（或其他介质）上传送，并被推送到你的应用程序，或由你的应用程序从任何外部来源获取。

无论您在程序中使用哪种类型的数据，都应知道，一切都皆可做流来观察，即使是静态数据和在程序中看起来是静态的数据也不例外。例如，配置数据被认为是静态的，即使是配置也会在某个时点发生改变，可能需要很长时间，也可能很短时间。从程序的角度来看，这并不重要，关键是在发生变化时做出响应并处理。如果将静态数据视为另一种数据流，就可以轻松地静态数据和数据流这两种类型的数据结合起来。对于应用程序来说，数据从哪里来并不重要。
例如，程序启动时通常会从持久化存储中加载数据，以恢复其配置（上次程序关闭前保存的配置）。当然，这种配置会在程序运行期间发生变化。在程序的内部，会去监听携带配置信息的数据流。当程序启动时，数据流将发送已加载的数据，而当配置发生变化时，数据流将发送更新。

在解释数据流时，我喜欢用水管来做类比，但这根水管中流动的是数据，就像图 1.16 中的水管一样。使用水管时，可以做很多事情。你可以在末端安装过滤器。您可以添加不同的接头，使其具有不同的功能。您可以添加压力监测器，帮助您调节流量。对数据流也可如此。您需要建立一个管道，管道里可以让信息在其中流动，并让其满足你的处理逻辑；处理逻辑包括过滤、转换、分组、合并等。

数据流和事件流非常适合用Rx的Observable(可观察序列)对象表达。可以使用 IObservable 接口对数据流和事件流进行抽象，然后构建嵌套的处理器(operator)，或者组装复杂执行流水线(前文叫管道,这里改叫流水线 对应英文Pipeline)。 这与在 Shoppy 示例中的做法类似，那个示例中，我们调用服务器的API获取折扣，而这个调用本身，就是复杂执行流水线的一部分，并且流水线还使用了过滤器（用来判断连接是否在线），并最终刷新了视图（就像洒水器洒水一样）。

1.6 总结

本章介绍了反应式的含义，以及如何使用 Rx 在应用程序中实现反应式编程。

• 在反应式编程中，你需要使用时变变量，这些变量的值会随着其依赖关系的变化而变化。您在 Shoppy 示例中看到了这些变量的例子：位置、连接、图标大小等。
• Rx 是微软开发的一个库，用于在.NET 应用程序中实现反应式编程。
• 在 Rx 中，时变变量被抽象成Observable(观察序列)，由IObservable 接口实现。
• Observable(观察序列)是通知的生产者，Observer(观察者)去订阅Observable(观察序列)，以接收这些通知。
• 每个Observer(观察者)在订阅后，都会返回一个IDisposable接口对象，允许随时取消订阅。
• Observer(观察者)用 IObserver 接口实现。
• Observable(可观察序列)可推送带有有效负载的通知、完成通知和错误通知。
• 在Observable(可观察序列)向Observer(观察者)发出完成通知或错误通知后，Observable(可观察序列)将不再发出通知。
• Observable(可观察序列)并不总是会完成或者完毕；它们也可以持续不断地发出通知。
• Observable(可观察序列)可以是 “安静的”，这意味着它们从未推送过任何元素，也永远不会推送。
• Rx 提供的处理器可用于创建查询、转换、投影等执行流水线，其语法与 LINQ 使用的语法相同。
• 弹珠图用于可视化 Rx 管道。
• 反应式系统的定义是具有响应性、弹性、伸缩性且采用消息驱动。反应式系统的这些特征在《反应式宣言》（Reactive Manifesto）中有所定义。
• 在反应式系统中，Rx 属于消息驱动模块，是一种处理程序接收消息的方式。
• 异步性是反应式系统最重要的部分之一，因为它可以让你更好地利用资源，从而使程序更灵敏。
• "万物皆流 "解释了为什么 Rx 可以轻松处理任何源，即使是数据库等数据源。
  在下一章中，你将有机会构建自己的第一个 Rx 程序，并与采用传统事件处理方式编写的相同程序进行比较。
  你将亲眼目睹 Rx 有多棒。