C# 如何实现一个观察者模式 - IObservable和IObserver

C#中观察者模式的官方实现是IObservable和IObserver接口,它们自.NET Framework 4.0起作为Rx基础契约,比手写事件更规范、可组合,并支持取消与错误传播。

和iobserver">

在 C# 中,观察者模式的官方实现就是 IObservableIObserver 接口,它们是 .NET Framework 4.0 起内置的响应式扩展(Rx)基础契约,比手写事件或自定义接口更规范、可组合、支持取消和错误传播。

核心角色与职责

IObservable 是“被观察者”,负责管理订阅关系、推送数据。它只暴露一个 Subscribe(IObserver) 方法,返回 IDisposable 用于取消订阅。
IObserver 是“观察者”,定义三个回调方法:
- OnNext(T value):接收新数据
- OnError(Exception error):接收异常通知(之后不再调用其他方法)
- OnCompleted():通知正常结束(之后不再调用其他方法)

手动实现一个简单 IObservable

适合理解原理,比如封装一个随时间推移产生数字的序列:

  • 继承 IObservable,内部维护观察者列表(线程安全建议用 ConcurrentBag>
  • Subscribe 中添加观察者,并返回一个 IDisposable 实现,用于从列表中移除该观察者
  • Task.RunTimer 模拟异步推送,在合适时机遍历观察者调用 OnNextOnCompletedOnError

注意:调用 OnNext/OnError/OnCompleted 前必须确保观察者不为 null,且每个观察者只能收到最多一次 OnErrorOnCompleted —— 这是契约关键。

更推荐:用 System.Reactive(Rx.NET)构造

手动实现易出错,实际开发中应优先使用 System.Reactive NuGet 包提供的工厂方法:

  • Observable.Range(1, 5) → 推送 1~5
  • Observable.Interval(TimeSpan.FromSeconds(1)) → 每秒推送一个 long 计数
  • Observable.FromEventPattern()将 .NET 事件转为可观测序列
  • Observable.Create(observer => { ... return () => { /* 取消逻辑 */ }; }) → 最灵活的手动构造方式,自动处理订阅/取消/异常捕获

例如:var source = Observable.Create(o => { o.OnNext("hello"); o.OnCompleted(); return Disposable.Empty; });

订阅与资源清理

调用 Subscribe 返回 IDisposable,务必妥善管理生命周期:

  • UI 场景中,在页面/控件卸载时调用 Dispose() 防止内存泄漏
  • 可用 using 语句(仅适用于同步短生命周期场景)
  • Rx 提供 CompositeDisposable 管理多个订阅,方便统一释放
  • 避免在 OnNext 中执行耗时或阻塞操作,否则会拖慢整个链路;必要时用 ObserveOn(Scheduler.ThreadPool) 切换线程

基本上就这些。用好 IObservableIObserver 的关键是理解“推送契约”和“生命周期责任”,而不是堆砌语法。Rx 的强大在于组合能力(WhereSelectSwitch 等),但底层仍是这两个接口在工作。