004.可观察对象与观察者

Rx非常适合事件驱动的应用程序。这是有意义的，因为事件(作为)(如前所述)是创建时变值的命令式方法。从历史上看,事件驱动编程主要出现在客户端技术中，因为作为事件实现的用户交互。例如，你可能工作过使用OnMouseMove或OnKeyPressed事件。正因如此，难怪你看到许多使用Rx的客户端应用程序。此外，一些客户端框架是基于Rx，如ReactiveUI (http://reactiveui.net)。

但是让我向您保证，Rx不是仅限客户端的技术。相反，服务器端代码的许多场景都非常适合Rx。另外，就像我说的在此之前，Rx用于大型应用程序，如Microsoft Cortana、Netflix和使用Microsoft StreamInsight的复杂事件处理(CEP)。Rx是一个优秀的库处理应用程序接收消息,而不管它在服务层或客户端层上运行。

接下来介绍什么是可观察对象。可观察对象（Observables）用于实现时变值(我们定义为可观察对象)序列)在Rx。它们代表推送模型（ push model ），在该模型中，新数据被推送到(或)通知)观察者。

可观察对象被定义为事件(或通知)的来源，或者，如果你愿意，也可以定义为数据流的发布者。推送模型意味着，不是让观察者从源中获取数据，而是在数据可用时将其传递给观察者，并总是检查是否有尚未获取的新数据（拉取模型）。

Observable实现IObservable接口，该接口位于.NET4.0的System空间里。

public interface IObservable<T>
{
 IDisposable Subscribe(IObserver<T> observer); 
}

IObservable接口只有一个方法Subscribe，它允许观察者订阅通知。Subscribe方法返回一个Disposable对象，该对象表示订阅，并允许观察者通过调用Dispose方法随时取消订阅。Observables保存订阅的观察者的集合，并在有值得通知的事情时通知他们。这是使用IObserver＜T＞接口完成的，该接口也位于.NET4.0的System命名空间。如图所示。

public interface IObserver<T>
{
 void OnNext(T value); 
 void OnError(Exception error); 
 void OnCompleted(); 
}

使用IObservable和IObserver的基本流程如图1.6所示。可观察对象并不总是完整的;它们可以是可能无限数量的序列元素(例如无限集合)的提供者。一个可观察对象也可以是“安静”，意思是它从来没有推动过任何元素，也永远不会。可观察对象也可以失败;失败可以发生在可观察对象已经推送了元素之后，也可以在没有任何元素被推动的情况下发生。

这个可观察的代数用下面的表达式形式化(其中*表示零次或多次，?表示零或一次，且“|”为“或”运算符):
在这里插入图片描述
图1.6是被观察者与观察者交互流程。在这个场景中，观察者被订阅到可观察对象应用程序;可观察对象向观察者“推送”三条消息(只有一条)在这种情况下)，然后通知观察者它已经完成。

当失败时，将使用OnError方法通知观察器，并使用异常对象将被传递给观察者进行检查和处理(参见
图1.7)。在出现错误之后(以及在完成之后)，将不会有更多的消息推给观察员。当观察者不提供错误处理程序时，Rx使用的默认策略是升级异常并导致崩溃。您将在第10章学习如何优雅地处理错误。
在这里插入图片描述
图1.7在可观察对象中出现错误的情况下，观察者将通过带有异常对象的OnError方法得到通知。

观察者设计模式

在某些编程语言中，事件有时作为一等公民提供，这意味着您可以使用所提供的语言定义和注册事件关键字和类型，甚至将事件作为参数传递给函数。
对于不支持事件作为一等公民的语言，使用观察者模式是一个有用的设计模式，允许您向应用程序添加类似事件的支持。此外，.NET的事件就是基于这种模式现实的。
观察者模式是由Gang of Four (GoF)在《设计模式》中引入的:
可重用的面向对象软件的要素(Addison-Wesley Professional, 1994)。该模式定义了两个组件:主体和观察者(不要混淆)(Rx的观察者)。观察者是对事件感兴趣的参与者将自身订阅到引发事件的主题。这是它在统一中的样子建模语言(UML)类图:

观察者模式很有用，但有几个问题。观察者只有一个方法来接受事件。如果要附加到多个主题或多个
除了一个事件，您还需要实现更多的更新方法。另一个问题是该模式没有指定处理错误的最佳方法，如果有的话，要由开发人员自己寻找通知错误的方法。最后要解决的问题是如何知道主题什么时候完成，Rx IObservable和IObserver基于Observer设计模式，但对其进行了扩展以解决这些问题的缺点。