Part 2 核心思想

《Rx.NET in Action》这一部共分八章，涵盖了Rx 关键模块——**Observable(可观察序列)和Observer(观察者)**的全部功能，以及如何创建它们、连接它们和控制它们之间的关系。

然后，您将学习如何使用强大的 Rx 处理器构建复杂的 Rx 管道。您将学习使用处理器查询单个Observable(可观察序列)，或查询多个Observable(可观察序列)的组合。您将了解如何在查询中控制并发性和时间参数化，以及如何在设计中处理故障和避免已知陷阱。

4 创建 Observable(可观察序列)

本章内容包括

• 创建数据和事件的 Observable(可观察序列)
• 从枚举对象创建 Observable(可观察序列)
• 使用 Rx 创建处理器
• 初步了解 Observable(可观察序列)

当人们开始学习 Rx 时，通常会问："我该从哪里开始呢？答案很简单：应该从创建 Observable(可观察序列) 开始。

在接下来的两章中，你将学习创建 Observable(可观察序列)的各种方法。本章仅限于创建同步的可观察序列(Observable)。第 5 章将介绍Observable(可观察序列)在异步创建和释放过程中涉及到的问题 。

由于存在多种类型的源，您可以从中接收项目，因此便不止一种方法来创建 Observable(可观察序列)。例如，您可以从传统的 .NET 事件中创建一个 Observable(可观察序列)，这样您仍然可以重用现有的代码；也可以从集合中创建 Observable(可观察序列)，这样更容易将它与其他 Observable(可观察序列)结合起来。每种方法都适用于不同的场景，并具有不同的意义，如简洁性和可读性。

4.1 使用 Observable(可观察序列)创建数据流和事件流

IObservable 接口是 Rx 最基本的构建模块，它只包含一个方法： Subscribe（订阅）。

Observable(可观察序列) 是推送出项目的源，另一端是接收项目的Observer(观察者)。项目可以有多种形式：它们可以是某件事情发生的通知（事件），也可以是可处理的数据元素（如聊天信息）。

图 4.1 显示了一个表示聊天应用程序接收到的聊天信息流的 Observable(可观察序列)。聊天Observer(观察者)通过 OnNext 方法接收每条消息，并可将其显示在屏幕上或保存到数据库中。在某一时刻，网络断开会导致错误通知。

图 4.1 Observable(可观察序列)与观察者之间可能发生的对话示例。Observable(可观察序列) 订阅后，观察者会收到通知，直到网络断开连接，从而导致出错。

我们将讨论几种获得这类 Observable(可观察序列)的方法。我们先从最简单的方法开始。

4.1.1 实现 IObservable 接口

在进入复杂的聊天程序示例之前，让我们先看看清单 4.1。

创建观察对象的最简单也是最幼稚的方法：手动实现 IObservable 接口。以这种方式创建观察序列并不是最好的做法，但我认为这对于理解观察序列的工作机制至关重要。

清单4.1 纯手工开发，推送数字的Observable(可观察序列)

using System;
using System.Reactive.Disposables;
public class NumbersObservable : IObservable<int>
{
    private readonly int _amount;
    public NumbersObservable(int amount)//将Observable(可观察序列)定义成：推送int给观察者
    {
        _amount = amount;
    }
    public IDisposable Subscribe(IObserver<int> observer)
    {
        for (int i = 0; i < _amount; i++)
        {
            observer.OnNext(i); //对于每个订阅的观察者，Observable(可观察序列) 都会推送一系列值。
        }
        observer.OnCompleted(); //Observable(可观察序列) 在推送所有值后通知结束。
        return Disposable.Empty; //Subscribe方法返回订阅对象，用来后期注销订阅用。
    }
}    

NumbersObservable 类实现了 IObservable 接口，允许任何整数观察者订阅该接口。请注意，当观察者订阅 NumbersObservable 时，NumbersObservable 会立即同步推送整数值。我们稍后将讨论使异步执行的 Observable(可观察序列)。

下面是一个观察者的示例，它将伴随我们走完本章。该观察者会在 OnNext、OnComplete 和 OnError 动作发生时将其写入控制台。

清单 4.2 ConsoleObserver将结果打印到控制台的观察者

public class ConsoleObserver<T> : IObserver<T> //订阅任何 Observable(可观察序列)并打印其发出的所有通知
{
    private readonly string _name;
    public ConsoleObserver(string name="")//在每个通知中打印名称（如果提供），以便于调试
    {
        _name = name;
    }
    public void OnNext(T value)//打印每个通知
    {
        Console.WriteLine("{0} - OnNext({1})",_name,value);
    }
    public void OnError(Exception error)//打印错误通知
    {
        Console.WriteLine("{0} - OnError:", _name); 
        Console.WriteLine("\t {0}", error);
    }
    public void OnCompleted()//打印完结通知
    {
        Console.WriteLine("{0} - OnCompleted()", _name);
    }
}    

下面显示了如何将 ConsoleObserver 订阅到 NumbersObservable：

var numbers = new NumbersObservable(5);
var subscription =
    numbers.Subscribe(new ConsoleObserver<int>("numbers"));

运行代码片段，您将看到以下内容：

numbers - OnNext(0) 
numbers - OnNext(1) 
numbers - OnNext(2) 
numbers - OnNext(3) 
numbers - OnNext(4) 
numbers - OnCompleted()

Observable推送给Observer五个数字，显示在带有 OnNext(n)的行中，在Observable完成之后，调用OnCompleted 。

每当Observer订阅了Observable，Observer就会收到一个 IDisposable 接口的对象。该对象保存着Observer的订阅，因此可以随时调用 Dispose 方法取消订阅。在我们这个发送一系列数字的简单示例中，Observable(可观察序列) 和Observer之间的整个通信都是在 Subscribe 方法中完成的，当该方法结束时，两者之间的连接也随之结束。在这种情况下，子订阅对象并没有真正的权力去取消订阅，但为了保持契约的正确性，我们可以使用 Rx 静态属性 Disposable.Empty 来返回一个空订阅对象。

注意 附录 B 详细介绍了 Rx Disposables库。

您可以让订阅 ConsoleObserver 的操作更方便。现在的方法是创建一个实例并在每次需要时进行订阅，不如创建一个扩展方法来帮你完成这项工作。

清单 4.3 SubscribeConsole 扩展方法

public static class Extensions
{
    public static IDisposable SubscribeConsole<T>(
            this IObservable<T> observable,
            string name="")
    {
        return observable.Subscribe(new ConsoleObserver<T>(name)); 
    }
}

在本书中，SubscribeConsole 将始终为您提供帮助，它可能对您的 Rx 测试和调查有用，因此是一个很好的工具。上一个示例现在看起来是这样的:

var numbers = new NumbersObservable(5);
var subscription = numbers.SubscribeConsole();

你现在已经手工创建了一个 Observable(可观察序列)和一个 Observer(观察者)，而且很容易。那为什么不能一直这样做呢？

4.1.2 手工编写 Observable(可观察序列)的问题

手工编写 Observable(可观察序列)是可行的，但很少使用，因为每次需要一个Observable时，就需要创建一个新的类型，既麻烦又容易出错。例如，Observable和Observer 关系规定，当 OnCompleted 或 OnError 被调用后，将不再向Observer(观察者)推送通知。如果更改 NumbersObservable，并在 OnComplete 被调用后再添加一次对观察者 OnNext 方法的调用，您就会发现它被调用了：

public IDisposable Subscribe(IObserver<int> observer)
{
    for (int i = 0; i < _amount; i++) 
    {
        observer.OnNext(i);
    }
    observer.OnCompleted();

    observer.OnNext(_amount); //observer 将收到通知
    return Disposable.Empty;
}

现在，这段代码将使 ConsoleObserver 输出如下内容：

errorTest - OnNext(0) 
errorTest - OnNext(1) 
errorTest - OnNext(2) 
errorTest - OnNext(3) 
errorTest - OnNext(4) 
errorTest - OnComplete
errorTest - OnNext(5)

这是有问题的，因为Observable和Observer之间的协议，才允许你创建 Rx 的各种处理器。例如，Repeat 处理器会在Observable完成时重新订阅Observer。如果Observable在完成过程中撒谎，那么使用 Repeat 的代码就会变得不可预测和混乱。

4.1.3 ObservableBase

除非某些特殊情况，我们一般不手动编写 Observable(可观察序列)。例如，你想命名Observable并让它封装复杂的逻辑时，那么适合手工编写。假设我们给Observer的每个方法和聊天通知（Received，Closed，Error）建立一种映射关系(后面与聊天服务通讯时会看到)，在此基础上，将聊天服务器的功能封装到一个类，该类为不同类型的通知（Received，Closed，Error）提供事件（event）。在这种情况下，我们希望通过一个推送聊天信息的 **Observable(可观察序列)**来使用聊天服务。连接到聊天服务时，我们会得到IChatConnection接口的连接对象：

//聊天连接接口定义了三种事件,所有聊天服务器产生的信号、消息、异常等，最终将转换成三种事件进行触发.
public interface IChatConnection
{
    event Action<string> Received;//当聊天消息抵达时,发送消息
    event Action Closed;//当聊天关闭时,发送消息
    event Action<Exception> Error;//当出错时,发送消息
    void Disconnect();
}    

与聊天服务的连接是通过聊天客户端类的 Connect 方法完成的：

public class ChatClient
{
    ...
    public IChatConnection Connect(string user, string password) 
    {
        // 连接 chat 服务
    }
}

我们用 Observable(可观察序列)来负责聊天信息的推送。如图 4.2 所示，每种聊天事件对应Observer的一个方法，形成我们之前所说的映射关系：

• Received 事件可映射到Observer的 OnNext
• Closed事件可与Observer的 OnComplete 映射
• Error 事件可映射到Observer的 OnError。

从映射可以看出，我们需要编写自己的逻辑去对应聊天事件，并最终调用到Observer的方法，因此比较合适创建自己的Observable类型。但是你肯定希望手动创建Observable时，尽可能少的产生的错误，因此 Rx 团队提供了一个基类： ObservableBase。下面的列表展示了如何使用它来创建 ObservableConnection 类。

清单 4.4 ObservableConnection

using System;
using System.Reactive;
using System.Reactive.Disposables;
public class ObservableConnection : ObservableBase<string>
{
    private readonly IChatConnection _chatConnection;
    public ObservableConnection(IChatConnection chatConnection)
    {
        _chatConnection = chatConnection; //保存连接对象，以便注册或取消事件监听
    }

    //ObservableBase 类提供了抽象方法 SubscribeCore，你可在其中编写订阅观察者的逻辑。
    protected override IDisposable SubscribeCore(IObserver<string> observer) 
    {
        //创建ChatConnection的三种事件处理程序。
        //将它们分别保存在received,closed,error三个Action类型的变量中，以便以后可以取消注册。
        Action<string> received = message =>
        {
            observer.OnNext(message);
        };
        Action closed = () =>
        {
            observer.OnCompleted();
        };
        Action<Exception> error = ex =>
        {
            observer.OnError(ex);
        };
        //注册ChatConnection的三种事件监听
        _chatConnection.Received += received; 
        _chatConnection.Closed += closed; 
        _chatConnection.Error += error;

        return Disposable.Create(() =>
        {//注销订阅时，取消ChatConnection的事件监听，并尝试断开与服务的连接。
            _chatConnection.Received -= received; 
            _chatConnection.Closed -= closed; 
            _chatConnection.Error -= error; 
            _chatConnection.Disconnect();
        });
        //译者注:整个函数看得我目瞪口呆,因为所有逻辑全部都写在一个函数内
        //作者首先，将事件处理函数封装在三个Action中
        //然后,将Action挂载到_chatConnection进行监听处理
        //最后，在取消订阅的时候,把三个Action从监听上拿下来。
        //这种对Action和闭包的使用简直是神来之笔。
        //将丛生到死的逻辑写的一目了然,看着确实舒服
    }
}        

提示 ObservableConnection 示例基于 SignalR 创建可观察连接的方式。SignalR 是一个帮助服务器端代码向连接的客户端推送内容的库。它功能强大，你应该去看看。

ObservableConnection 派生于 ObservableBase，并实现了抽象方法 SubscribeCore，该方法最终会被 ObservableBase Subscribe 方法调用。ObservableBase 会对观察者进行有效性检查（比如是否为空），并执行观察者和Observable之间的契约。为此，它将每个观察者封装在一个名为 AutoDetachObserver 的封装器中。当观察者调用 OnCompleted 或 OnError 或观察者本身在接收消息时抛出异常时，该封装器会自动将观察者从客户端分离。这就减轻了在Observable中，自己实现这一安全执行管道的负担。

获得 ObservableConnection 后，就可以对其进行订阅：

var chatClient = new ChatClient();
var connection = chatClient.Connect("guest", "guest"); //连接聊天服务器

//从 ChatConnection 创建 ObservableConnection。尚未订阅。
IObservable<string> observableConnection = new ObservableConnection(connection);

//利用本章开头的ConsoleObserver观察者进行订阅，注意这里使用扩展方法的方式进行订阅
var subscription= observableConnection.SubscribeConsole("receiver");

和以前一样，您通过扩展方法，可以让创建 ObservableConnection 的过程更加轻松愉快。

public static class ChatExtensions
{
    public static IObservable<string> ToObservable( 
            this IChatConnection connection)
    {
        return new ObservableConnection(connection); 
    }
}

现在，你只需这样写即可：

var subscription = 
    chatClient.Connect("guest", "guest") 
    .ToObservable() 
    .SubscribeConsole();//译者注：再次目瞪口，方法链，领域语言，简洁。我已经说都不会话了。

尽管如此，每次创建新的Observable类型还是很烦人，而且大多数情况下你也没有这么复杂的逻辑需要维护。这就是为什么直接从 ObservableBase 或 IObservable 接口派生创建Observable被认为是不好的做法。相反，你应该使用 Rx 库提供的现有工厂方法之一来创建 Observable(可观察序列)。

译者注：总结就是八个字 ，直接不好，使用工厂。本小节信息量很大，最终翻译完毕，撒花！（好像不是花，是头发）

4.1.4 使用 Observable.Create 创建Observable(可观察序列)

每个可观察序列都实现了 IObservable 接口，但您不需要手动完成。位于 System.Reactive.Linq 命名空间下的静态类型 Observable 提供了几个静态方法来帮助您创建Observable对象。Observable.Create方法允许你通过传递订阅方法的代码来创建Observable。下面的列表展示了如何使用它来创建之前手动创建的数字可观察序列。

清单 4.6 使用 Observable.Create 创建数字可观察序列

Observable.Create<int>(observer =>
{
    for (int i = 0; i < 5; i++)
    {
        observer.OnNext(i);
    }
    observer.OnCompleted();
    return Disposable.Empty;
});

与之前使用的 ObservableBase 一样，Create 方法会为你完成所有的模板工作。它会创建一个 Observable 实例——类型为 AnonymousObservable，并将你提供的委托（在本例中为 lambda 表达式）附加到 Observable Subscribe 方法中。

Observable.Create则更进一步，它不仅允许你返回创建的 IDisposable，还允许你返回 Action。所提供的 Action 可以容纳您的注销订阅代码，在它返回后，Create 方法将把 Action 包裹在一个通过 Disposable.Create 创建的 IDisposable 对象中。如果返回null，Rx 将为您创建一个空的Disposable。

注意 附录 B 详细介绍了 Rx Disposables库。

当然，您希望创建的 Observable(可观察序列)可以自定义元素个数（上一示例中为 5 个）。如图所示，我们可以定义一个方法去创建 Observable(可观察序列)：

public static IObservable<int> ObserveNumbers(int amount)
{
    return Observable.Create<int>(observer =>
    {
        for (int i = 0; i < amount; i++)
        {
            observer.OnNext(i);
        }
        observer.OnCompleted();
        return Disposable.Empty;
    });
}

Observable.Create因其灵活易用而被大量使用，但您可能希望将可观察序列的创建推迟到需要的时候，例如观察者订阅时。

4.1.5 推迟创建 Observable(可观察序列)

在第 4.1.2 节中，我们使用 ChatClient 类连接到远程聊天服务器，然后将返回的 ChatConnection 转换为Observable对象，将消息推送到观察者中。连接服务器和将连接转换为可观察序列这两个步骤总是同时进行的，因此我们希望在 ChatClient 中添加 ObserveMessages 方法，对其进行封装，并遵循 "不要重复自己"原则(DRY——Don’t Repeat Yourself) ：

public IObservable<string> ObserveMessages(string user, string password)
{
    var connection = Connect(user, password);//此处代码会立即连接chat服务器
    return connection.ToObservable();
}

每当调用 ObserveMessages 方法时，都会创建一个聊天服务连接，然后将其转换为一个 Observable(可观察序列)。这样做效果很好，但有可能在创建Observable对象后，很长时间都没有观察者订阅该Observable，或者从来没有观察者订阅该Observable。出现这种情况的一个原因是，您可能会创建一个 Observable(可观察序列)并将其传递给其他方法或对象，而这些方法或对象可能会在自己的时间内使用它（例如，一个界面只有在加载时才会去订阅，但只有当父视图接收到用户输入时才会加载）。

然而，连接是打开的，会浪费你的机器和服务器上的资源。倒不如将连接延迟到观察者订阅的那一刻。这就是图 4.3 所示Observable.Defer 处理器的作用。

图 4.3 Defer 方法的签名

Defer（延迟的意思）处理器会创建一个观察序列，作为真实Observable的代理。当观察者订阅时，参数中 observableFactory 函数将被调用，观察者将订阅所创建的Observable。此序列如图 4.4 所示。

图 4.4 观察者订阅时，Defer 处理器创建Observable的序列图。

当你想使用工厂处理器（下面会学到）创建观察者，或者你有一个自己的工厂方法（你可能、或者不能或不想改变该方法），但你仍然想在观察者订阅时创建该观察者时，Defer 就是个好办法。

如下就是如何使用Defer来创建 ObservableConnection：

public IObservable<string> ObserveMessagesDeferred(string user, 
                                                   string password)
{
    return Observable.Defer(() =>
    {
        var connection = Connect(user, password); 
        return connection.ToObservable();
    });
}

我需要指出的是，使用 Defer 并不意味着多个观察者可以共享用 observableFactory 创建的Observable对象。如果有两个观察者订阅了从 Defer 方法返回的 Observable，Connect 方法就会被调用两次：

var messages = chatClient.ObserveMessagesDeferred("user","password"); 
var subscription1 = messages.SubscribeConsole();
var subscription2 = messages.SubscribeConsole();//结果是再次呼叫 "Connect"。

这种行为并非Defer 所特有的。同样的问题也出现在你之前创建的 Observable(可观察序列)中。请牢记这一点，在第 6 章谈到冷Observable和热Observable时，你将了解何时以及如何制作可共享的Observable。Defer 在观测变量的 "温度 "世界中还扮演着另一个角色，因为它可以用来将热观测变量变为冷观测变量，不过我说得有点远了。

最终，你创建的 Observable(可观察序列)会将传统的 .NET 事件桥接到了 Rx 中。这是您在使用 Rx 时经常要做的事情，因此 Rx 提供的处理器可以减轻这项工作。

4.2 从事件创建Observable(可观察序列 )

从传统的 .NET 事件创建可观察序列，这在前面的章节中你已经看到过，但我们还没有讨论过里面会发生什么。如果您需要的只是将传统的 .NET 事件转换为可观察对象，那么使用 Observable.Create 等方法就显得多余了。相反，Rx 提供了两种将事件转换为可观察序列的方法，即 FromEventPattern 和 FromEvent。这两种方法（或处理器）经常会让使用 Rx 的人员感到困惑，因为使用错误的方法会导致编译错误或异常。

4.2.1 创建符合 EventPattern 的 Observable(可观察序列)

您在.NET 框架中看到的.NET 事件处理程序具有以下签名：

void EventHandler(object sender, DerivedEventArgs e)

事件处理程序一般会接收一个sender对象（即事件的发送者）和一个派生自 EventArgs 类的事件参数对象。由于传递发送者和事件参数的模式非常常用，因此您可以在 .NET Framework 中找到创建事件时可以使用的通用委托： EventHandler 和 EventHandler。

Rx 意识到使用这种结构的委托来创建事件很常见，也就是所谓的事件模式，因此提供了一种方法来轻松转换遵循事件模式的事件。这就是 FromEventPattern 处理器。FromEventPattern 有几个重载，最常用的重载如图 4.5 所示。

图 4.5 FromEventPattern 方法重载签名之一

addHandler 和 removeHandler 参数非常有趣。因为这两个参数指定如何附加和分离 事件处理程序TDelegate。要求以Action类型提供，Action中需要处理注册TDelegate和注销TDelegate的过程。在大多数情况下（不是全部）都是用固定的lambda编写，代码格式如下：

• addHandler： h => [src].[event] += h 表示监听 [src].[*event]*件事，发生事件时调用函数h进行处理
• removeHandler： h => [src].[event] -= h 表示取消监听，[src].[event]发生时，不再调用h函数

例如，在 WPF 等UI事件中就可以看到事件模式。假设你想接收窗口中名为 theButton 的按钮的单击事件。Button 类公开了 Click 事件（该事件在其基类 ButtonBase 中定义）：

public event RoutedEventHandler Click;

其中的RoutedEventHandler 是一个委托，定义在 System.Windows 命名空间中：

public delegate void RoutedEventHandler(object sender, 
                                        System.Windows.RoutedEventArgs e)

要从单击事件创建一个 Observable(可观察序列)，则编写以下代码：

IObservable<EventPattern<RoutedEventArgs>> clicks = 
           FromEventPattern<RoutedEventHandler, RoutedEventArgs>( 
                    h => theButton.Click += h,
                    h => theButton.Click -= h);
clicks.SubscribeConsole();//将信息写入 Visual Studio 输出窗口

在这里，我们将 Click 事件转换为Observable(可观察序列)，这样，每个事件发生时都会调用观察者的 OnNext 方法。 RoutedEventHandler 是事件定义中指定的事件处理程序， RoutedEventArgs是事件发送给事件处理程序的事件参数类型。

创建的 Observable(可观察序列) 会推送 EventPattern 类型的对象。该类型封装了发送者和事件参数的值。

如果使用标准的 EventHandler 来定义事件，则可以使用 FromEventPattern 重载，该重载只需要 TEventArgs 的通用参数：

IObservable<EventPattern<TEventArgs>> FromEventPattern<TEventArgs>( 
     Action<EventHandler<TEventArgs>> addHandler, 
     Action<EventHandler<TEventArgs>> removeHandler);

Rx 还提供了将事件转换为 Observable(可观察序列)的最简单版本，其中只需指定事件名称(字符串)和持有该事件的对象，因此点击事件示例可以写成下面这样：

IObservable<EventPattern<object>> clicks = 
     Observable.FromEventPattern(theButton, "Click");

我不喜欢这种方法。"Click"这种魔力字符串容易造成各种错误和代码混乱 ，而且很容易打错字，如果您决定重新命名事件，还需要记住更改应用程序中的字符串。不过，这种简单的方法还是很有吸引力的，因此请谨慎使用。

提示 如果您正在处理 GUI 应用程序，并发现 UI 事件到 Observable(可观察序列)之间的转换很有吸引力，那么您可能会发现 ReactiveUI 框架 (http://reactiveui.net/) 很有帮助。本书没有涉及 ReactiveUI，但它提供了许多有用的 Rx 工具。其中之一就是许多 UI 事件到 Observable(可观察序列)的内置转换。

4.2.2 不遵循事件模式的事件

并非所有事件都遵循事件模式。假设有一个扫描某区域可用 Wi-Fi 网络的类。当某个网络可用时，该类就会触发一个事件：

public delegate void NetworkFoundEventHandler(string ssid); 
class WifiScanner
{
    public event NetworkFoundEventHandler NetworkFound = delegate { }; 
    // rest of the code
}

该事件不符合标准事件模式，因为事件处理程序需要接收一个字符串（网络 SSID）。如果尝试使用 FromEventPattern 方法将事件转换为 Observable(可观察序列)，就会出现参数异常，因为 NetworkFoundEventHandler 委托不能转换为标准的 EventHandler 类型。

为了解决这个问题，Rx 提供了与 FromEventPattern 方法类似的 FromEvent 方法：

IObservable<TEventArgs> FromEvent<TDelegate, TEventArgs> 
    (Action<TDelegate> addHandler, Action<TDelegate> removeHandler);

FromEvent 方法的重载需要两个泛型参数：一个是委托TDelegate——用来附加到事件的，另一个是事件参数类型TEventArgs——事件传递给委托用的。这里最重要的一点是，对委托TDelegate或事件参数TEventArgs没有任何限制；它们可以是你喜欢的任何类型。如下是 WifiScanner 类的编写方法：

var wifiScanner = new WifiScanner();
IObservable<string> networks = 
    Observable.FromEvent<NetworkFoundEventHandler, string>( 
            h => wifiScanner.NetworkFound += h,
            h => wifiScanner.NetworkFound -= h);

在代码中，我们根据 WifiScanner 公开的事件，创建了一个Observable(可观察序列)。该事件需要一个 NetworkFoundEventHandler 委托类型的事件处理程序，而事件处理程序接收的值是字符串，因此生成的观察序列是 IObservable<string> 类型。

4.2.3 带有多个参数的事件

如果Observable(可观察序列)中事件需要多个参数，那么 EventHandler 的方法签名中就需要有多个参数，此时就需要使用其他FromEvent重载方法，不过会复杂一些。例如，WifiScanner 不仅要发送网络名称，还要发送网络强度：

//事件处理程序需要接收 SSID 和信号强度。
public delegate void ExtendedNetworkFoundEventHandler(string ssid, 
                                                      int strength);
class WifiScanner
{
    public event ExtendedNetworkFoundEventHandler ExtendedNetworkFound = delegate { };
}

此时，再用与单参数版本的代码是行不通的。在调用观察者的 OnNext 方法时，Observable(可观察序列) 只能传递一个值。您需要以某种方式将两个参数封装在一个对象中。所以门使用带有转换函数的 FromEvent 重载，将 Rx 事件处理程序转换为可与事件一起使用的事件处理程序。首先，让我们来看看方法签名，然后我会解释你看到的具体内容。

IObservable<TEventArgs> FromEvent<TDelegate, TEventArgs> (
    Func<Action<TEventArgs>, TDelegate> conversion, 
    Action<TDelegate> addHandler,
    Action<TDelegate> removeHandler
);
//其中conversion是转换函数。该函数的传参是Action<TEventArgs>。这就是将调用观察者 OnNext 的 Rx 处理程序。
//然后，转换处理程序会传递给 addHandler 和 removeHandler。

这个方法签名需要时间来消化，因此我们通过一个示例来解释它，以帮助理解。下面的示例将 ExtendedNetworkFound 事件转换为 Tuple<string, int> 类型的Observable(可观察序列)：

IObservable<Tuple<string, int>> networks = Observable.FromEvent<ExtendedNetworkFoundEventHandler, Tuple<string, int>>( 
    //创建一个转换函数，将ssid和strengthTuple封装在Tuple中。
    rxHandler =>
        (ssid, strength) =>  rxHandler(Tuple.Create(ssid, strength)),
    h => wifiScanner.ExtendedNetworkFound += h ,
    h => wifiScanner.ExtendedNetworkFound -= h);

首先让我们来谈谈 addHandler 和 removeHandler。与之前一样，这两个Action会接收一个方法的引用，该方法用来负责事件监听的相关事宜。addHandler 注册该方法，而 removeHandler 则取消注册该方法。但 Rx 如何知道要创建什么handler呢？这就是conversion的工作。在示例中，conversion就是 lambda 表达式，具体如下：

//实例中conversion的lambda 
rxHandler =>(ssid, strength) => rxHandler(Tuple.Create(ssid, strength))
    
//FromEvent函数对conversion的定义
Func<Action<TEventArgs>, TDelegate> conversion 

示例中TEventArgs是Tuple<string, int>，TDelegate是ExtendedNetworkFoundEventHandler。
转换后，可以理解为：

Func<Action<Tuple<string, int>>, ExtendedNetworkFoundEventHandler> conversion  

然后结合lambda代码，进行分析。rxHandler 其实是 Action<Tuple<string, int> ,是conversion的传参，所以conversion看起来应该是这样的一个函数：

ExtendedNetworkFoundEventHandler conversion(Action<Tuple<string, int> rxHandler)

(ssid, strength) =>…这段代码 ,其实是conversion的返回值,即返回一个ExtendedNetworkFoundEventHandler委托类型的函数,注意是返回一个函数或者说一个委托,如果我们给这个函数起个名字叫retFun,则代码如下:

void retFun (string ssid,int strength)    

=>rxHandler(Tuple.Create(ssid, strength) 这段代码就比较难理解了,其实这就是retFun的函数体：

void retFun (string ssid,int strength){
    rxHandler(Tuple.Create(ssid, strength));
}

难点就在rxHandler是一个传入的参数，而且还是个函数型参数，然后我们在retFun内，调用了这个函数。所以整个lambda，如果掰开揉碎了来看，应该理解为：

ExtendedNetworkFoundEventHandler conversion(Action<Tuple<string, int> rxHandler){
    ExtendedNetworkFoundEventHandler retFun=>(string ssid,int strength){
        rxHandler(Tuple.Create(ssid, strength));
    };
    return retFun;
}

这个 lambda 表达式的目的是创建一个处理函数（conversion），这个处理函数负责返回一个与事件的委托定义相匹配的委托实例（说白了就是返回一个函数），代换到示例中可以理解为，lambda创建了一个处理函数，该处理函数返回了一个ExtendedNetworkFoundEventHandler类型的委托实例，而ExtendedNetworkFoundEventHandler类型有两个传参ssid和strength。但是该处理函数的传数却是Action<Tuple<string, int>>(还是一个函数，也就是说传参是函数，返回值还是函数，所以需要仔细理解)。然后lambda 表达式将该处理函数的传参取名 rxHandler。最后lambda表达式用**=>rxHandler(Tuple.Create(ssid, strength))**告知编译器说，我定义一个匿名委托函数作为返回值，其中匿名委托函数的代码是rxHandler(Tuple.Create(ssid, strength))。而最后的最后Tuple<string, int>类型的数据，会在调用观察者的OnNext函数中，作为传参进行传递。如此，便实现了事件期望与 Rx 期望之间的中介。

译者注：这一小节lambda的使用非常精彩，但是在英文原文中解释lambda时，写的晦涩难懂。这与每个人对抽象事物的理解力有关，我的想法是尽可能用通俗直白的语句描述问题，哪怕这种表达不严谨，但至少能看懂。而作者的表达方式更加严谨，但是也特别抽象。

就比如作者说"如此，便实现了事件期望与 Rx 期望之间的中介"，虽然这句话我保留了直译，但是要我自己表达，会说“如此，便实现了将事件中的多参（ssid, strength）封装成Rx可以使用的单参（Tuple）”。我这样表达明显更加直白。而作者表达的则更加抽象。所以这一章节中，我加了很多自己的解释、说明以及代码，这些并非来自原文。但确有助于理解多参处理函数的工作原理。

总结来说呢，在处理多参事件时，需要一个转换后函数，该函数的传参是函数，返回值还是函数。但是不管怎么弯弯绕，最终就是把多参包装到一个参数里面，因为OnNext就支持一个参数。别问为什么，问就是规定！

4.2.4 处理无参数的事件

并非每个事件都会向其事件处理程序发送参数。某些事件说明发生了某些事情；例如，当网络连接时，WiFiScanner 会引发下一个事件：

event Action Connected = delegate { };

当尝试将事件转换为可观察序列时，会遇到一个问题。每个可观察序列都必须实现 IObservable，而 T 是将推送给观察者的数据类型。那么从 Connected 事件创建的 T 类型是什么呢？我们需要一个可以表示空的中性类型。在数学中，与运算（如乘法或加法）相关的中性元素被称为 “Unit”（它实际上被称为 “Identity”，但在更广泛的上下文中被称为 “Unit”）。单位（Unit）元素在日常生活中已经非常熟悉：在乘法运算中是数字 1，在加法运算中是数字 0。这就是 Rx 包含 System.Reactive.Unit 结构的原因。该结构体没有真正的成员，您可以将其视为代表单一值的空实体。它通常用来表示一个 void 返回方法的成功完成，就像我们的事件一样。下面介绍如何将事件转换为 Observable(可观察序列)：

IObservable<Unit> connected = Observable.FromEvent( 
    h => wifiScanner.Connected += h,
    h => wifiScanner.Connected -= h);
connected.SubscribeConsole("connected");

由于 Unit 只有一个默认值（代表 void），因此它的 ToString 方法会返回空括号"() "字符串，所以从前面的示例中可以得到以下输出结果：

connected - OnNext(()) 
connected - OnNext(())

注意 我们没有介绍 FromEventPattern 和 FromEvent 的一些重载。通过这些重载，您可以简化简单情况下的事件挂钩（例如，事件处理程序只是 Action），或将不符合事件模式的事件转换为 EventPattern 的 IObservable。您应该看一看。

将事件转换为可观察序列很有帮助，因为有了可观察序列后，使用 Rx 处理器对事件进行处理就没有任何限制了。但事件并不是唯一可以转换成可观察序列的构造；有时，你会想把与可观察序列完全相反的东西转换成可观察序列。我说的就是Enumerable(可枚举序列)。

4.3 从Enumerable(可枚举序列)到 Observable(可观察序列) 或者反向转换

Enumerable可提供在拉动模型中工作的机制，而Observable可让您在推动模型中工作。有时你会想从拉式模型转到推式模型，以创建两个世界的标准处理方式，例如创建相同的逻辑来添加即时接收到的聊天信息和存储并随后从存储库中读取的信息。有时，从推送模式转移到拉动模式甚至是有意义的。本节将探讨这些转换及其对代码的影响。

4.3.1 Enumerable(可枚举序列)到Observable(可观察序列)

Enumerable(可枚举序列)和Observable(可观察序列)是对偶的；通过几个简单的步骤，就可以从一个变量转换为另一个变量。

可以通过几个简单的步骤将Enumerable变量转换为Observable变量。Rx 提供了一种方法，可以帮助您将可枚举序列转换为可观察序列： ToObservable 方法。在本示例中，您将创建一个字符串数组并将其转换为可观察序列：

IEnumerable<string> names = new []{"Shira", "Yonatan", "Gabi", "Tamir"}; IObservable<string> observable = names.ToObservable();
observable.SubscribeConsole("names");

代码输出如下：

names - OnNext(Shira) 
names - OnNext(Yonatan) 
names - OnNext(Gabi) 
names - OnNext(Tamir) 
names - OnCompleted()

ToObservable 方法会创建一个观察序列，一旦子订阅该观察序列，观察序列就会对集合进行迭代，并将每个项目传递给观察者。迭代完成后，观察序列将调用 OnComplete 方法。

如果在迭代过程中出现异常，将把异常传递给 OnError 方法。

class Program
{
    static void Main(string[] args)
    {
        NumbersAndThrow() 
            .ToObservable() 
            .SubscribeConsole("names");
        Console.ReadLine(); 
    }
    static IEnumerable<int> NumbersAndThrow()
    {
        yield return 1;
        yield return 2;
        yield return 3;
        throw new ApplicationException("Something Bad Happened"); 
        yield return 4;
    }
}    

该示例的输出结果如下：

enumerable with exception - OnNext(1) 
enumerable with exception - OnNext(2) 
enumerable with exception - OnNext(3) 
enumerable with exception - OnError:
	System.ApplicationException: Something Bad Happened . . .

如果只需要最终订阅枚举，则可以在枚举上使用订阅扩展方法。这会将 enumerable 转换为 Observable(可观察序列) 并对其进行订阅：

IEnumerable<string> names = new[] { "Shira", "Yonatan", "Gabi", "Tamir" }; names.Subscribe(new ConsoleObserver<string>("subscribe"));

在何处使用

在本章开头，我们创建了一个 ObservableConnection，它通过一个观察序列来消费聊天信息。ObservableConnection 的本质是客户端只接收新消息，但作为用户，你希望进入聊天室查看连接前的消息。

为了简化我们的方案，我们假设在您离线时没有发送任何消息。现在我们需要加载之前会话中保存的消息。通常，这就是需要用到数据库。程序会将收到的每条信息保存到数据库中，当您连接时，这些信息就会被加载并添加到屏幕上。

有了 ObservableConnection，且也已经有了如何将消息添加到屏幕上的代码。现在我们需要从数据库加载消息的代码。如果能将数据库中的消息表示为一个可观察序列(observable)，并将其与新消息的可观察序列(observable)合并，然后使用同一个观察者来接收来自两个世界的消息，那就更好了。下面的小示例就能做到这一点：合并两条被保存到数据库中的消息，和两条在连接时被接收的消息：

ChatClient client = new ChatClient();
IObservable<string> liveMessages = client.ObserveMessages("user","pass"); 
IEnumerable<string> loadedMessages = LoadMessagesFromDB();
loadedMessages.ToObservable() 
    .Concat(liveMessages) 
    .SubscribeConsole("merged");

本例使用了Concat处理器 。该处理器将把 liveMessages 观察序列连接到 loadedMessages 观察序列，这样，只有在所有加载的信息都发送给观察者后，才会发送实时信息。输出结果如下:

merged - OnNext(loaded1) 
merged - OnNext(loaded2) 
merged - OnNext(live message1)
merged - OnNext(live message2)

也可以如下这样编写，而无需转换Enumerable：

liveMessages 
    .StartWith(loadedMessages) 
    .SubscribeConsole("loaded first");

StartWith 处理器首先向观察者发送Enumerable中的所有值，然后开始发送在 liveMessages 上收到的所有消息。

在前面几章中，我们讨论了Enumerable/Observable的对偶性，你可以看到它们允许双向转换：从Enumerable到Observable，这里已经看到；以及从Observable到Enumerable，接下来你将看到。

4.3.2 Observable 到 Enumerable

与将enumerable转换为Observable 的方法相同，您也可以使用 ToObservable相反的方法，使用 ToEnumerable 方法。这将创建一个Enumerable，一旦遍历该Enumerable，就会阻塞线程，直到出现可用项或观察者序列(Observable)完成为止。使用 ToEnumerable 可能都是些情非得已的情况，比如我们某些现有的库代码只接受Enumerable，而我们需要用该代码遍历 Observable(可观察序列)中的已有项目，例如，按时间或按数量对项目的一部分进行排序。使用 ToEnumerable 非常简单，这里演示一下。

Listing 4.8 Using the ToEnumerable operator

var observable = 
    Observable.Create<string>(o =>
    {
        o.OnNext("Observable"); 
        o.OnNext("To"); 
        o.OnNext("Enumerable"); 
        o.OnCompleted();//如果注释这一行，线程将在 OnNext 中的所有值被消费后进入等待状态。
        return Disposable.Empty;
    });
var enumerable = observable.ToEnumerable();
foreach (var item in enumerable) //循环将打印 OnNext 发送的每个值。当 Observable(可观察序列) 完成时，循环将结束。
{
    Console.WriteLine(item);
}

由于 ToEnumerable 返回的enumerable存在阻塞行为，因此不建议使用它。应尽量使用推送模型。

注意 Next 处理器也返回一个enumerable，但它的操作方式与 ToEnumerable不同。第 6 章将介绍这一主题。

Rx 包含能以非阻塞方式将 Observable 转换为列表和数组（保持其为Observable ）的方法，分别是 ToList 和 ToArray。与 ToEnumerable 不同，这些方法返回的Observable只包含一个元素（如果出现错误，则不提供任何值），即一个List(列表)或一个Array(数组)。列表（或数组）只有在观察序列完成时才会发送给观察者。

清单4.9 使用ToList处理器

var observable =
    Observable.Create<string>(o =>
    {
        o.OnNext("Observable"); 
        o.OnNext("To");
        o.OnNext("List");
        o.OnCompleted();//只有当 Observable(可观察序列)完成时，列表才会发送给观察者。
        return Disposable.Empty;
    });
IObservable<IList<string>> listObservable = observable.ToList();
listObservable
    .Select(lst => string.Join(",", lst))  //将列表转换为字符串，其中每个项目都用逗号分隔。
    .SubscribeConsole("list ready");

运行该示例的结果如下:

list ready - OnNext(Observable,To,List) 
list ready - OnCompleted()

本着将 Observable(可观察序列)转换为enumerable的精神，我还应该提及 ToDictionary 和 ToLookup 方法。虽然听起来很相似，但它们的用例却各不相同。

将 Observable(可观察序列)转换为字典

在 .NET 中，实现接口 System.Collections.Generic .IDictionary<TKey, TValue> 的类型被称为包含键值对的类型。对于每个键，只能有一个对应的值或根本没有值。在这种情况下，我们说键不是字典的一部分。

Rx 提供了一种将 Observable(可观察序列)转化为字典的方法，即使用ToDictionary 方法，该方法存在几种重载。下面的示例是最简单的一个：

IObservable<IDictionary<TKey, TSource>> ToDictionary<TSource, TKey>( 
     this IObservable<TSource> source,
     Func<TSource, TKey> keySelector)

该方法对源Observable推送的每个值运行 keySelector，并将其添加到字典中。源Observable完成后，字典就会发送给 Observer(观察者)。下面的小示例演示了如何根据城市名称创建字典，其中的键是名称的长度。

清单 4.10 使用ToDictionary处理器

IEnumerable<string> cities = new[] { "London", "Tel-Aviv", "Tokyo", "Rome" };
var dictionaryObservable =
    cities
    .ToObservable()
    .ToDictionary(c => c.Length);//Key可以根据你需要，但如果两个项目共享相同的Key，就会出现异常。
dictionaryObservable
    .Select(d => string.Join(",", d)) 
    .SubscribeConsole("dictionary");

运行该示例会显示如下内容：

dictionary - OnNext([6, London],[8, Tel-Aviv],[5, Tokyo],[4, Rome]) 
dictionary - OnCompleted()

如果 Observable(可观察序列)中的两个值共享同一个键，那么在尝试将它们添加到字典时，就会收到一个异常，提示该键已经存在。字典的键和值之间必须保持 1:1 的关系；如果你希望每个键有多个值，就需要LOOKUP(查找器)。

将 Observable(可观察序列)转换为LOOKUP(查找器)

如果需要将 Observable(可观察序列) 转换为类似字典的结构，每个键保存多个值，那么 ToLookup 就是你所需要的。ToLookup 的签名与 ToDictionary 的签名相似：

IObservable<ILookup<TKey, TSource>> ToLookup<TSource, TKey>(
     this IObservable<TSource> source, Func<TSource, TKey> keySelector)

与 ToDictionary 一样，您需要为每个 Observable(可观察序列)指定键（其他重载也允许您指定值本身）。您可以将LOOKUP(查找器)看作一个词典，只是其中每个值都是一个集合。

下一个示例创建了一个由城市名称组成的 Observable(可观察序列)，其中的键是名称的长度。这次，Observable(可观察序列) 将有多个名称长度相同的城市。

Listing 4.11 Using the ToLookup operator

IEnumerable<string> cities =
     new[] { "London", "Tel-Aviv", "Tokyo", "Rome", "Madrid" };
var lookupObservable = 
    cities
    .ToObservable()
    .ToLookup(c => c.Length);
lookupObservable
    .Select(lookup =>
    {
        var groups = new StringBuilder();
        foreach (var grp in lookup) 
            groups.AppendFormat("[Key:{0} => {1}]",grp.Key,grp.Count()); 
        return groups.ToString();
    })
    .SubscribeConsole("lookup");

运行该示例会显示如下内容：

lookup - OnNext([Key:6 => 2][Key:8 => 1][Key:5 => 1][Key:4 => 1]) 
lookup - OnCompleted()

您可以看到，由于London和Madrid有相同的长度（6），输出结果显示键 6 有两个值。

Observable(可观察序列)和Enumerable之间的对偶性允许你在两个世界中进行操作，并使你可以很容易地根据需要将其中一个转换为另一个。但请记住，这也是一个警告。除了 "实现 "其逻辑或从其他类型转换，您还有更多方法来创建 Observable。常见的模式都很好地用创建处理器进行了抽象，并可用作工厂。

4.4 使用 Rx 创建处理器

到此为止，我们已经了解了如何手工创建 Observable(可观察序列)或从已知类型（如Enumerable和事件）进行转换。随着时间的推移，我们逐渐发现，可观察对象创建中的某些模式在不断重复，例如在循环内部发射项或发射一系列数字。与其自己编写，Rx 提供的处理器有助于以标准而简洁的方式完成这些工作。由创建处理器创建的可观察序列通常被用作更复杂的可观察序列的构建模块。

4.4.1 生成 Observable 循环

假设您需要运行一个类似于迭代的过程，让Observable 产生序列元素，批量读取文件一样。在这种情况下，可以使用 Observable.Generate处理器。下面是其最简单的重载：

IObservable<TResult> Generate<TState, TResult>(
    TState initialState,
    Func<TState, bool> condition,
    Func<TState, TState> iterate,
    Func<TState, TResult> resultSelector)

例如，如果您想生成一个观察序列，推送前 10 个偶数（从 0 开始），您可以这样做：

IObservable<int> observable =
    Observable.Generate(
        0,              //初始状态
        i => i < 10,    //条件 (false时结束) 
        i => i + 1,     //下移循环的步长处理
        i => i*2);      //生成每个循环元素
observable.SubscribeConsole();

运行此示例可打印出数字 0、2、4、6、8、10、12、14、16、18。

为了更简单起见，如果您要创建的是一个能创建元素范围的 Observable，您可以使用另一个仅能创建元素范围的处理器：Observable.Range 处理器：

IObservable<int> Range(int start, int count)

这将创建一个Observable，推送指定范围内的数字。

如果添加 Select 处理器，则可以创建与使用 Generate 创建的相同的 Observable(可观察序列)：

IObservable<int> observable =
    Observable
        .Range(0, 10)
        .Select(i => i*2);

Generate 或 Range 可用于创建更多的数字生成器。下面的示例使用 Generate 创建了一个观察序列(Observable)，它可以发出文件的行数。

4.4.2 读取文件

基本上，读取文件是一个反复的过程。您需要打开文件并逐行读取，直到读完为止。在Observable的世界里，你希望把内容推送给你的 Observable(可观察序列)。下面的代码展示了如何使用 Observable.Generate来实现这一功能：

IObservable<string> lines = 
    Observable.Generate( 
        File.OpenText("TextFile.txt"),
        s => !s.EndOfStream,
        s => s,
        s => s.ReadLine());
lines.SubscribeConsole("lines");

这是我在一个有四行的文件上运行示例时得到的结果：

lines - OnNext(The 1st line) 
lines - OnNext(The 2nd line) 
lines - OnNext(The 3rd line) 
lines - OnNext(The 4th line) 
lines - OnCompleted()

释放文件资源

上一个示例有一个你可能无法立即发现的缺陷。对 File.OpenText 的调用创建了一个打开文件的流。即使在 Observable(可观察序列)完成后–无论是当它到达终点时，还是当它从外部被处理掉时–流仍处于活动状态，文件仍处于打开状态。为了克服这一问题，并使您的应用程序能正确处理资源，您需要让 Rx 知道可释放资源的存在。这就是 Observable.Using处理器的作用所在。Using 处理器会接收创建资源的工厂方法（以及用该资源创建Observable的工厂方法）。返回的 Observable 将确保内部 Observable 完成后，资源将被处置。

注意 使用处理器以及其他资源管理注意事项将在第 10 章中介绍。
本清单展示了如何修正我们的示例。

清单 4.12 使用 Using 处理器释放资源

IObservable<string> lines = 
    Observable.Using(
        () => File.OpenText("TextFile.txt"), 
        stream => 
            Observable.Generate( 
                stream,
                s => !s.EndOfStream,
                s => s,
                s => s.ReadLine())
        );

现在你可以确定，当你的 observable用过后，你创建的任何资源都不会保持未注销状态，从而使你的代码更加高效。

4.4.3 原始Observable(可观察序列)

在某些情况下，一些创建处理器可以派上用场，与其他observable相结合，创建边缘情况。这不仅在测试或演示和学习时有用，而且在自己建立处理器，处理某些需要特殊处理的输入时也很有用。

创建单项 Observable(可观察序列)

Observable.Return（可观察序列）处理器用于创建一个可观察对象，向观察者推送一个单项，然后完成：

Observable.Return("Hello World") 
    .SubscribeConsole("Return");

运行此示例的结果如下：

Return - OnNext(Hello World) 
Return - OnCompleted()

创建永不结束的 Observable(可观察序列)

Observable.Never(可观察序列)用于创建一个不向观察者推送任何项目且永远不会完成（甚至不会出错）的Observable：

Observable.Never<string>() 
    .SubscribeConsole("Never");

泛型参数用于确定 Observable(可观察序列)的类型。您也可以通过一个假值来完成同样的操作。运行此示例不会在屏幕上打印任何内容。

创建可抛出的 Observable(可观察序列)

如果需要模拟观察对象通知其观察者错误的情况，Observable.Throw 将帮助您实现这一目标：

Observable.Throw<ApplicationException>(
     new ApplicationException("something bad happened")) 
    .SubscribeConsole("Throw");

这是运行示例后打印的内容：

Throw - OnError:
         System.ApplicationException: something bad happened

创建空的 Observable(可观察序列)

如果需要一个不向观察者推送任何项目并立即完成的观察对象，可以使用 Observable.Empty 操作符：

 Observable.Empty<string>() 
    .SubscribeConsole("Empty");

输出如下：

Empty - OnCompleted()

4.5 小结

哇，你在本章中学到了很多。你应该为自己感到骄傲。本章所涉及的内容几乎会伴随你创建的每一个Observable流水线：

• 所有Observable都实现了 IObservable 接口。

• 不鼓励通过手动实现 IObservable 接口来创建 Observable(可观察序列)。相反，请使用内置的创建处理器之一。

• 通过 Create 处理器，您可以通过传递订阅方法来创建 Observable(可观察序列)，该方法将为每个订阅的观察者运行。

• Defer处理器允许您推迟或延迟创建Observable，直到有观察者订阅序列为止。

• 要从符合事件模式（其中使用的委托会接收发送者和 EventArgs）的事件中创建观察序列，请使用 FromEventPattern 处理器。

• 要转换不符合事件模式的事件，请使用 FromEvent 处理器。

• FromEventPattern 和 FromEvent 处理器接收一个向事件添加事件处理程序的函数，以及一个从事件移除事件处理程序的函数。

• 您可以使用 FromEventPattern处理器的重载，它允许您传递一个对象，并指定要创建Observable的事件名称。这主要用于标准框架事件。

• 使用处理器 ToObservable也可将Enumable对象Observable对象。

• 可以使用处理器 ToEnumerable、ToList、ToDictionary 和 ToLookup 将 Observable(可观察序列)转换为Enumable对象。但根处理器符的不同，它们会导致消耗代码阻塞，直到出现可用项或整个观察序列(observable)完成为止。

• 要从迭代过程中创建Observable，请使用 Generate（生成）处理器。

• Range（范围）处理器创建一个Observable。

• 要创建一个发出一个通知的Observable，请使用 Observable.Return 处理器。

• 要创建一个从不发出通知的观察对象，请使用 Observable.Never 处理器。

• 要创建一个可发出失败通知的观察对象，请使用 Observable.Throws处理器。

• 要创建一个空的观察对象，请使用 Observable.Empty 处理器。

尽管如此，在本章中，我们还是忽略了包装异步执行的重要类型。下一章将扩展你关于创建 Observable(可观察序列)的知识。您还将了解 .NET 中的异步模式以及如何将它们与 Rx.Observable.Empt 处理器连接起来。