C# IObservable與IObserver通知機制 觀察者模式（推式模型）

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
 
namespace ConsoleApp1
{
    class Program
    {
        static void Main(string[] args)
        {
            /*
                C# IObservable與IObserver通知機制 觀察者模式（推式模型）             
             */
            WeatherDataPublisher publisher = new WeatherDataPublisher();
            CurrentConditionDisplay currentDisplay = new CurrentConditionDisplay();
            StatisticsConditionDisplay statisticsDisplay = new StatisticsConditionDisplay();
            //訂閱當前天氣展示板
            IDisposable currentDisplayUnsubscriber = publisher.Subscribe(currentDisplay);
            //訂閱氣溫統計展示板
            IDisposable statisticsDisplayUnsubscriber = publisher.Subscribe(statisticsDisplay);
 
            for (int i = 0; ; i++)
            {
                WeatherData weatherData = new WeatherData();
                Console.WriteLine("請輸入溫度,濕度,壓力");
                string input = Console.ReadLine();
                var array = input.Split(',');
                weatherData.temperature = array[0];
                weatherData.humility = array[1];
                weatherData.pressure = array[2];
                Console.WriteLine("");
                //將輸入的新的天氣數據傳給天氣數據發布器
                publisher.ReciveNewData(weatherData);
                Console.WriteLine("=============================");
            }
        }
    }
 
    /// <summary>
    /// WeatherData類包含氣溫，濕度，氣壓等屬性。
    /// </summary>
    class WeatherData
    {
        /// <summary>
        /// 氣溫
        /// </summary>
        public string temperature { get; set; }
        /// <summary>
        /// 濕度
        /// </summary>
        public string humility { get; set; }
        /// <summary>
        /// 氣壓
        /// </summary>
        public string pressure { get; set; }
    }
 
    /// <summary>
    /// WeatherDataPublisher類實現了IObservable接口，實現了Subscribe訂閱方法。
    /// </summary>
    class WeatherDataPublisher : IObservable<WeatherData>
    {
        List<IObserver<WeatherData>> observers = new List<IObserver<WeatherData>>();
 
        /// <summary>
        /// 訂閱主題，將觀察者添加到列表中
        /// </summary>
        /// <param name="observer"></param>
        /// <returns></returns>
        public IDisposable Subscribe(IObserver<WeatherData> observer)
        {
            observers.Add(observer);
            return new Unsubscribe(this.observers, observer);
        }
 
        /// <summary>
        /// 取消訂閱類
        /// </summary>
        private class Unsubscribe : IDisposable
        {
            List<IObserver<WeatherData>> observers;
            IObserver<WeatherData> observer;
            public Unsubscribe(List<IObserver<WeatherData>> observers
            , IObserver<WeatherData> observer)
            {
                this.observer = observer;
                this.observers = observers;
            }
 
            public void Dispose()
            {
                if (this.observers != null)
                {
                    this.observers.Remove(observer);
                }
            }
        }
 
        /// <summary>
        /// 通知已訂閱的觀察者
        /// </summary>
        /// <param name="weatherData"></param>
        private void Notify(WeatherData weatherData)
        {
            foreach (var observer in observers)
            {
                observer.OnNext(weatherData);
            }
        }
 
        /// <summary>
        /// 接收最新的天氣數據
        /// </summary>
        /// <param name="weatherData"></param>
        public void ReciveNewData(WeatherData weatherData)
        {
            Notify(weatherData);
        }
    }
 
    /// <summary>
    /// 抽象類WeatherDisplayBase實現了IObserver接口，所有的天氣展示板（觀察者）繼承這個抽象類，
    /// 需實現OnNext方法，即接收到新數據推送后要進行的數據處理展示工作，並且可重寫OnCompleted，OnError方法。
    /// </summary>
    abstract class WeatherDisplayBase : IObserver<WeatherData>
    {
        public virtual void OnCompleted()
        {
        }
        public virtual void OnError(Exception error)
        {
        }
        public abstract void OnNext(WeatherData value);
    }
 
    /// <summary>
    /// CurrentConditionDisplay類為當前天氣狀況展示板，繼承WeatherDisplayBase抽象類，展示最新的天氣數據。
    /// </summary>
    class CurrentConditionDisplay : WeatherDisplayBase
    {
        public override void OnNext(WeatherData value)
        {
            Console.WriteLine("------------------");
            Console.WriteLine("當前天氣狀況板");
            Console.WriteLine(string.Format("溫度：{0}\n濕度：{1}\n氣壓：{2}",
                value.temperature, value.humility, value.pressure));
        }
    }
 
    /// <summary>
    /// StatisticsConditionDisplay類為氣溫統計展示板，繼承WeatherDisplayBase抽象類，展示歷史最高溫度，最低溫度，平均溫度。
    /// </summary>
    class StatisticsConditionDisplay : WeatherDisplayBase
    {
        List<float> temperatures = new List<float>();
        public override void OnNext(WeatherData value)
        {
            float temperature;
            if (float.TryParse(value.temperature, out temperature))
            {
                temperatures.Add(temperature);
            }
            Console.WriteLine("------------------");
            Console.WriteLine("溫度統計板");
            Console.WriteLine(string.Format("平均溫度：{0}\n最高溫度：{1}\n最低溫度：{2}",
                temperatures.Average(), temperatures.Max(), temperatures.Min()));
        }
    }
}

 

注解

很多時候被觀察者（IObservable）向觀察者（IObserver）提供的數據並不像Location這樣簡單的結構體。

而是一個包含復雜數據的類，通常可能是被觀察者本身，這種情況是允許的，即IObserver<T> 實現和 T 可以表示同一類型。

這時候的實現變成下面的型式：

public class LocationTracker2 : IObservable<LocationTracker>
{
    public IDisposable Subscribe(IObserver<LocationTracker> observer)
    {
        throw new NotImplementedException();
    }
}

public class LocationReporter2 : IObserver<LocationTracker2>
{
    public void OnCompleted()
    {
        throw new NotImplementedException();
    }

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

    public void OnNext(LocationTracker2 value)
    {
        throw new NotImplementedException();
    }
}

 觀察者模式是常用的設計模式，在.net環境下，其運行時庫為開發者提供了IObservable<T>和 IObserver<T>接口，用於實現觀察者模式軟件設計。

  //
    // 摘要:
    //     定義基於推送的通知的提供程序。
    //
    // 類型參數:
    //   T:
    //     提供通知信息的對象。
    public interface IObservable<out T>
    {
        //
        // 摘要:
        //     通知提供程序：某觀察程序將要接收通知。
        //
        // 參數:
        //   observer:
        //     要接收通知的對象。
        //
        // 返回結果:
        //     使資源釋放的觀察程序的接口。
        IDisposable Subscribe(IObserver<T> observer);
    }

注解Subscribe：

調用Subscribe通知提供程序某觀察程序將要接收通知（即注冊、訂閱），不同於常規實現，它具有一個返回值，是一個IDisposable對象，當觀察者不再接收通知時，可調用Dispose函數取消訂閱（反注冊），這種方法充分發揮C#語言的特性。

IObserver<in T>

//
// 摘要:
//     提供用於接收基於推送的通知的機制。
//
// 類型參數:
//   T:
//     提供通知信息的對象。
public interface IObserver<in T>
{
    //
    // 摘要:
    //     通知觀察者，提供程序已完成發送基於推送的通知。
    void OnCompleted();
    //
    // 摘要:
    //     通知觀察者，提供程序遇到錯誤情況。
    //
    // 參數:
    //   error:
    //     一個提供有關錯誤的附加信息的對象。
    void OnError(Exception error);
    //
    // 摘要:
    //     向觀察者提供新數據。
    //
    // 參數:
    //   value:
    //     當前的通知信息。
    void OnNext(T value);
}

 

示例

下面例子演示觀察者設計模式，實現定位系統實時通知當前經緯度坐標。

包含經緯度坐標的Locaiton結構體

public struct Location
{
    public Location(double latitude, double longitude)
    {
        Latitude = latitude;
        Longitude = longitude;
    }

    public double Latitude
    {
        get; private set;
    }

    public double Longitude
    {
        get;
        private set;
    }
}

LocationTracker 類
實現了IObservable<T> 接口。

public class LocationTracker : IObservable<Location>
{
    public LocationTracker()
    {
        observers = new List<IObserver<Location>>();
    }

    private List<IObserver<Location>> observers;

    public IDisposable Subscribe(IObserver<Location> observer)
    {
        if (!observers.Contains(observer))
            observers.Add(observer);
        return new Unsubscriber(observers, observer);
    }
    // 用於取消訂閱通知的IDisposable對象的實現
    private class Unsubscriber : IDisposable
    {
        private List<IObserver<Location>> _observers;
        private IObserver<Location> _observer;

        public Unsubscriber(List<IObserver<Location>> observers, IObserver<Location> observer)
        {
            this._observers = observers;
            this._observer = observer;
        }

        public void Dispose()
        {
            if (_observer != null && _observers.Contains(_observer))
                _observers.Remove(_observer);
        }
    }
    // TrackLocation 方法傳遞了一個包含緯度和經度數據的Location對象。 
    // 如果Location值不為null，則 TrackLocation 方法會調用每個觀察程序的 OnNext 方法，
    // 否則調用OnError方法
    public void TrackLocation(Nullable<Location> loc)
    {
        foreach (var observer in observers)
        {
            if (!loc.HasValue)
                observer.OnError(new LocationUnknownException());
            else
                observer.OnNext(loc.Value);
        }
    }

    public void EndTransmission()
    {
        foreach (var observer in observers.ToArray())
            if (observers.Contains(observer))
                observer.OnCompleted();

        observers.Clear();
    }
}

public class LocationUnknownException : Exception
{
   internal LocationUnknownException() 
   { }
}

LocationObserver類

定位信息的觀察者，實現了IObserver<Location>接口

public class LocationReporter : IObserver<Location>
{
    private IDisposable unsubscriber;
    private string instName;

    public LocationReporter(string name)
    {
        this.instName = name;
    }

    public string Name
    { get { return this.instName; } }

    public virtual void Subscribe(IObservable<Location> provider)
    {
        if (provider != null)
            unsubscriber = provider.Subscribe(this);
    }

    public virtual void OnCompleted()
    {
        Console.WriteLine("The Location Tracker has completed transmitting data to {0}.", this.Name);
        this.Unsubscribe();
    }

    public virtual void OnError(Exception e)
    {
        Console.WriteLine("{0}: The location cannot be determined.", this.Name);
    }

    public virtual void OnNext(Location value)
    {
        Console.WriteLine("{2}: The current location is {0}, {1}", value.Latitude, value.Longitude, this.Name);
    }
    // 取消訂閱
    public virtual void Unsubscribe()
    {
        unsubscriber.Dispose();
    }
}

 

最后來看一下怎么使用這個定位系統

class Program2
{
    static void Main(string[] args)
    {
        // Define a provider and two observers.
        LocationTracker provider = new LocationTracker();
        LocationReporter reporter1 = new LocationReporter("FixedGPS");
        reporter1.Subscribe(provider);
        LocationReporter reporter2 = new LocationReporter("MobileGPS");
        reporter2.Subscribe(provider);

        provider.TrackLocation(new Location(47.6456, -122.1312));
        reporter1.Unsubscribe();
        provider.TrackLocation(new Location(47.6677, -122.1199));
        provider.TrackLocation(null);
        provider.EndTransmission();
    }
}