基本概念

The Reactive Extensions (Rx) is a library for composing asynchronous and event-based programs using observable sequences and LINQ-style query operators.

Reactive Extensions represents all these data sequences as observable sequences. An application can subscribe to these observable sequences to receive asynchronous notifications as new data arrive.

Push-based/Pull-based

Rx的主要的接口

IObservable<T>/IObserver<T>

IObservable<T>就是observable sequences 的抽象，就像Pull-based中的IEnumerable<T>类似，用来承载数据，代表了一个可以被观察的数据源，他可以在将sequences中的数据，推送给任何一个有兴趣的listener，我们用IObserver<T>来代表这种对IObservable<T>感兴趣的listener。

一个IObservable<T>的实现可以被视为一个T类型数据的集合， 例如IObservable<int>可以被视为一个int类型数据的集合，他们会被推送给IObserver<T>这样的订阅者。

我们看一下接口的定义：

public interface IObservable<out T>
{
   IDisposable Subscribe(IObserver<T> observer);
}
public interface IObserver<in T>
{
   void OnCompleted();            // Notifies the observer that the source has finished sending messages.
   void OnError(Exception error); // Notifies the observer about any exception or error.
   void OnNext(T value);          // Pushes the next data value from the source to the observer.
}

Rx还通过扩展的方式提供了一些订阅更方便的方式，不需要实现自己的IObserver<T>，只需要实现对应的订阅事件(OnNext， OnError, OnComplete)对应的delegate方法即可，如下面的例子：

IObservable<int> source = Observable.Range(1, 5); //creates an observable sequence of 5 integers, starting from 1
IDisposable subscription = source.Subscribe(
                            x => Console.WriteLine("OnNext: {0}", x), //prints out the value being pushed
                            ex => Console.WriteLine("OnError: {0}", ex.Message),
                            () => Console.WriteLine("OnCompleted"));

上面这只是两个例子，在实际的开发应用中，我没几乎没有必要自己去实现这两个接口，Rx通过Observable和Observer为我们提供了丰富的足够的实现，后面将继续说明。

null