大概看了下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.Subscribe(this); } public virtual void Unsubscribe() { cancellation.Dispose(); flightInfos.Clear(); } public virtual void OnCompleted() { flightInfos.Clear(); } // No implementation needed: Method is not called by the BaggageHandler class. public virtual void OnError(Exception e) { // No implementation. } // Update information. public virtual void OnNext(BaggageInfo info) { bool updated = false; // Flight has unloaded its baggage; remove from the monitor. if (info.Carousel == 0) { var flightsToRemove = new List<string>(); string flightNo = String.Format("{0,5}", info.FlightNumber); foreach (var flightInfo in flightInfos) { if (flightInfo.Substring(21, 5).Equals(flightNo)) { flightsToRemove.Add(flightInfo); updated = true; } } foreach (var flightToRemove in flightsToRemove) flightInfos.Remove(flightToRemove); flightsToRemove.Clear(); } else { // Add flight if it does not exist in the collection. string flightInfo = String.Format(fmt, info.From, info.FlightNumber, info.Carousel); if (! flightInfos.Contains(flightInfo)) { flightInfos.Add(flightInfo); updated = true; } } if (updated) { flightInfos.Sort(); Console.WriteLine("Arrivals information from {0}", this.name); foreach (var flightInfo in flightInfos) Console.WriteLine(flightInfo); Console.WriteLine(); } } } using System; using System.Collections.Generic; public class Example { public static void Main() { //发布者 BaggageHandler provider = new BaggageHandler(); // 订阅者 ArrivalsMonitor observer1 = new ArrivalsMonitor("BaggageClaimMonitor1"); // 订阅者 ArrivalsMonitor observer2 = new ArrivalsMonitor("SecurityExit"); // 发布 provider.BaggageStatus(712, "Detroit", 3); //订阅 observer1.Subscribe(provider); provider.BaggageStatus(712, "Kalamazoo", 3); provider.BaggageStatus(400, "New York-Kennedy", 1); provider.BaggageStatus(712, "Detroit", 3); observer2.Subscribe(provider); provider.BaggageStatus(511, "San Francisco", 2); provider.BaggageStatus(712); observer2.Unsubscribe(); provider.BaggageStatus(400); provider.LastBaggageClaimed(); } }