大概看了下C#官方提供的IObservable接口以及IObserver接口来实现发布和订阅,写的很标准,很有代表性,做下笔记,以后要是项目需要用到发布订阅再基于自己的需求改:
public class BaggageInfo { private int flightNo; private string origin; private int location; internal BaggageInfo(int flight, string from, int carousel) { this.flightNo = flight; this.origin = from; this.location = carousel; } public int FlightNumber { get { return this.flightNo; } } public string From { get { return this.origin; } } public int Carousel { get { return this.location; } } } /// /// 发布者 /// public class BaggageHandler : IObservable<BaggageInfo> { private List<IObserver<BaggageInfo>> observers; private List<BaggageInfo> flights; public BaggageHandler() { observers = new List<IObserver<BaggageInfo>>(); flights = new List<BaggageInfo>(); } public IDisposable Subscribe(IObserver<BaggageInfo> observer) { // Check whether observer is already registered. If not, add it if (! observers.Contains(observer)) { observers.Add(observer); // Provide observer with existing data. foreach (var item in flights) observer.OnNext(item); } return new Unsubscriber<BaggageInfo>(observers, observer); } // Called to indicate all baggage is now unloaded. public void BaggageStatus(int flightNo) { BaggageStatus(flightNo, String.Empty, 0); } public void BaggageStatus(int flightNo, string from, int carousel) { var info = new BaggageInfo(flightNo, from, carousel); // Carousel is assigned, so add new info object to list. if (carousel > 0 && ! flights.Contains(info)) { flights.Add(info); foreach (var observer in observers) observer.OnNext(info); } else if (carousel == 0) { // Baggage claim for flight is done var flightsToRemove = new List<BaggageInfo>(); foreach (var flight in flights) { if (info.FlightNumber == flight.FlightNumber) { flightsToRemove.Add(flight); foreach (var observer in observers) observer.OnNext(info); } } foreach (var flightToRemove in flightsToRemove) flights.Remove(flightToRemove); flightsToRemove.Clear(); } } public void LastBaggageClaimed() { foreach (var observer in observers) observer.OnCompleted(); observers.Clear(); } } internal class Unsubscriber<BaggageInfo> : IDisposable { private List<IObserver<BaggageInfo>> _observers; private IObserver<BaggageInfo> _observer; internal Unsubscriber(List<IObserver<BaggageInfo>> observers, IObserver<BaggageInfo> observer) { this._observers = observers; this._observer = observer; } public void Dispose() { if (_observers.Contains(_observer)) _observers.Remove(_observer); } } /// /// 订阅者 /// public class ArrivalsMonitor : IObserver<BaggageInfo> { private string name; private List<string> flightInfos = new List<string>(); private IDisposable cancellation; private string fmt = "{0,-20} {1,5} {2, 3}"; public ArrivalsMonitor(string name) { if (String.IsNullOrEmpty(name)) throw new ArgumentNullException("The observer must be assigned a name."); this.name = name; } public virtual void Subscribe(BaggageHandler provider) { cancellation = provider.Sub