Reactive Extensions简介一

在.Net 4.0中引入了两个新的接口用来实现观察者模式——IObservable和IObserver。IObservable是数据源，IObserver是观察者，观察者订阅数据源后，当新的数据产生时，将其主动传给所有的订阅者（Iobserver）。

观察者模式比较基础，因此在这里并不多加介绍，没有相关基础的朋友可以参看MSDN的这两个链接。

l Observer（观察器）

l 探究观察者设计模式

这里我用一个简单的实例介绍一下这个接口的基本用法。

    class Program

    {

        static void Main(string[] args)

        {

            var timerServer = new TimeServer();

            timerServer.Subscribe(new Watch());

 

            System.Threading.Thread.Sleep(-1);

        }

    }

 

    class TimeServer : IObservable<DateTime>

    {

        public TimeServer()

        {

            new System.Threading.Timer(_ => Notify(DateTime.Now), null, 0, 1000);

        }

 

        void Notify(DateTime time)

        {

            foreach (var observer in _observers)

            {

                observer.OnNext(time);

            }

        }

 

        #region IObservable<DateTime> 成员

 

        List<IObserver<DateTime>> _observers = new List<IObserver<DateTime>>();

 

        public IDisposable Subscribe(IObserver<DateTime> observer)

        {

            //这里省略了参数有效性检查

            _observers.Add(observer);

            return new AnonymousUnSubscriber() { Action = () => _observers.Remove(observer) };

        }

 

        #endregion

 

        #region AnonymousUnSubscriber类

 

        class AnonymousUnSubscriber : IDisposable

        {

            public Action Action { get; set; }

 

            void IDisposable.Dispose()

            {

                this.Action();

            }

        }

 

        #endregion

    }

 

    class Watch : IObserver<DateTime>

    {

        #region IObserver<DateTime> 成员

 

        public void OnCompleted() { throw new NotImplementedException(); }

        public void OnError(Exception error) { throw new NotImplementedException(); }

 

        public void OnNext(DateTime value)

        {

            Console.WriteLine(value);

        }

 

        #endregion

    }

这里定义了两个简单的对象：TimeServer是数据源，实现了IObservable接口，Watch是观察者，实现了IObserver接口。当有新数据产生时（定时器每秒钟通知一次），会调用Watch的OnNext接口，将当前时间在屏幕上打印出来。

像这种“推”的方式一般也被称作反应式（Reactive），虽然其接口比较简明，但在.Net 4.0中只有接口的声明，并没有其它的相关库函数进行支持。因此要实现反应式编程得自己实现所有其它的相关代码。不过好在微推出了一个名为Reactive Extensions for .NET (Rx)的库，实现了许多强有力的功能，使得我们能快速构建强大的反应式的程序。像上面的这个例子，用Rx库可以简化如下： 

    static void Main(string[] args)

    {

        var timerServer = Observable.Interval(TimeSpan.FromSeconds(1)).Select(_ => DateTime.Now);

        timerServer.Subscribe(i => Console.WriteLine(i));

 

        System.Threading.Thread.Sleep(-1);

    }

 

Rx库非常强大，由于相关资料不多，目前我也是在学习和摸索中，后面还会写一些文章陆续介绍这个库。