使用IObserver/IObservable實現觀察者和主題

Question

我想要創建一個可用於表示動態計算值的類，而另一個表示值的類可以是這些動態計算值的源（主題）。目標是當主題改變時，計算值自動更新。

在我看來，使用IObservable/IObserver是要走的路。不幸的是，我不能使用Reactive Extensions庫，所以我不得不從頭開始實現主題/觀察者模式。

夠布拉布拉，這裏有我的課：http://msdn.microsoft.com/en-us/library/dd990377.aspx：

public class Notifier<T> : IObservable<T> 
{ 
    public Notifier(); 
    public IDisposable Subscribe(IObserver<T> observer); 
    public void Subscribe(Action<T> action); 
    public void Notify(T subject); 
    public void EndTransmission(); 
} 

public class Observer<T> : IObserver<T>, IDisposable 
{ 
    public Observer(Action<T> action); 
    public void Subscribe(Notifier<T> tracker); 
    public void Unsubscribe(); 
    public void OnCompleted(); 
    public void OnError(Exception error); 
    public void OnNext(T value); 
    public void Dispose(); 
} 

public class ObservableValue<T> : Notifier<T> 
{ 
    public T Get(); 
    public void Set(T x); 
} 

public class ComputedValue<T> 
{ 
    public T Get(); 
    public void Set(T x); 
} 

我實現從主要是解除。

那麼這樣做的「正確」方法是什麼？注意：我不關心LINQ或多線程甚至性能。我只是想讓它變得簡單易懂。

Answer 1

如果我是你，我會嘗試儘可能接近Rx的方式來實現你的類。

其中一個關鍵的基本原則是使用相對較少的具體類，這些類使用大量操作進行組合。所以你應該創建一些基本的構建模塊並使用組合將它們結合在一起。

在Reflector.NET下我將首先看兩個類：AnonymousObservable<T> & AnonymousObserver<T>。特別是AnonymousObservable<T>被用作實例化observables的基礎。實際上，如果您查看IObservable<T>派生的對象，有幾個專門的實現，但只有AnonymousObservable<T>用於通用目的。

靜態方法Observable.Create<T>()實質上是AnonymousObservable<T>的包裝。

明顯符合您的要求的其他Rx類是BehaviorSubject<T>。受試者既是可觀察者又是觀察者，因爲它記住了接收到的最後一個值，所以它符合你的情況。

鑑於這些基本的類，那麼你幾乎擁有創建特定對象所需的所有位。您的對象不應該從上面的代碼繼承，而是使用合成將您需要的行爲集合在一起。現在

，我會建議一些改變你的類的設計，使他們其中Rx更兼容，從而更加composible和魯棒性。

我會放棄你的Notifier<T>班，轉而使用BehaviourSubject<T>。

我會放棄你的Observer<T>班，轉而使用AnonymousObserver<T>。

然後，我會修改ObservableValue<T>看起來像這樣：

public class ObservableValue<T> : IObservable<T>, IDisposable 
{ 
    public ObservableValue(T initial) { ... } 
    public T Value { get; set; } 
    public IDisposable Subscribe(IObserver<T> observer); 
    public void Dispose(); 
} 

的ObservableValue<T>的實施將包裹BehaviourSubject<T>而不是繼承它暴露出IObserver<T>成員將允許訪問OnCompleted & OnError這不會使因爲這個類代表一個價值而不是一個計算，所以這個意義太大了訂閱將使用AnonymousObservable<T>和Dispose將清理包裝的BehaviourSubject<T>。

然後，我會修改ComputedValue<T>看起來像這樣：

public class ComputedValue<T> : IObservable<T>, IDisposable 
{ 
    public ComputedValue(IObservable<T> source) { ... } 
    public T Value { get; } 
    public IDisposable Subscribe(IObserver<T> observer); 
    public void Dispose(); 
} 

的ComputedValue<T>類將包裝AnonymousObservable<T>爲所有用戶和並用source搶值的本地副本爲Value財產。 Dispose方法將用於取消訂閱source觀測值。

最後兩個類是您的設計似乎需要的唯一真正的具體實現 - 這只是因爲Value屬性。

接下來，你需要一個靜態ObservableValues類的擴展方法：

public static class ObservableValues 
{ 
    public static ObservableValue<T> Create<T>(T initial) 
    { ... } 

    public static ComputedValue<V> Compute<T, U, V>(
     this IObservable<T> left, 
     IObservable<U> right, 
     Func<T, U, V> computation) 
    { ... } 
} 

的Compute方法將使用AnonymousObservable<V>執行計算併產生IObservable<V>傳遞到由返回的ComputedValue<V>構造方法。

帶着所有這些地方，你現在可以這樣寫代碼：

var ov1 = ObservableValues.Create(1); 
var ov2 = ObservableValues.Create(2); 
var ov3 = ObservableValues.Create(3); 

var cv1 = ov1.Compute(ov2, (x, y) => x + y); 
var cv2 = ov3.Compute(cv1, (x, y) => x * y); 

//cv2.Value == 9 

ov1.Value = 2; 
ov2.Value = 3; 
ov3.Value = 4; 

//cv2.Value == 20 

請讓我知道這是有益的和/或如果有什麼我可以詳細闡述。

---

編輯：還需要一些一次性使用。

您還需要實現AnonymousDisposable & CompositeDisposable管理特別是在Compute擴展方法訂閱。看看使用Reflector.NET的Rx實現或使用下面的我的版本。

public sealed class AnonymousDisposable : IDisposable 
{ 
    private readonly Action _action; 
    private int _disposed; 

    public AnonymousDisposable(Action action) 
    { 
     _action = action; 
    } 

    public void Dispose() 
    { 
     if (Interlocked.Exchange(ref _disposed, 1) == 0) 
     { 
      _action(); 
     } 
    } 
} 

public sealed class CompositeDisposable : IEnumerable<IDisposable>, IDisposable 
{ 
    private readonly List<IDisposable> _disposables; 
    private bool _disposed; 

    public CompositeDisposable() 
     : this(new IDisposable[] { }) 
    { } 

    public CompositeDisposable(IEnumerable<IDisposable> disposables) 
    { 
     if (disposables == null) { throw new ArgumentNullException("disposables"); } 
     this._disposables = new List<IDisposable>(disposables); 
    } 

    public CompositeDisposable(params IDisposable[] disposables) 
    { 
     if (disposables == null) { throw new ArgumentNullException("disposables"); } 
     this._disposables = new List<IDisposable>(disposables); 
    } 

    public void Add(IDisposable disposable) 
    { 
     if (disposable == null) { throw new ArgumentNullException("disposable"); } 
     lock (_disposables) 
     { 
      if (_disposed) 
      { 
       disposable.Dispose(); 
      } 
      else 
      { 
       _disposables.Add(disposable); 
      } 
     } 
    } 

    public IDisposable Add(Action action) 
    { 
     if (action == null) { throw new ArgumentNullException("action"); } 
     var disposable = new AnonymousDisposable(action); 
     this.Add(disposable); 
     return disposable; 
    } 

    public IDisposable Add<TDelegate>(Action<TDelegate> add, Action<TDelegate> remove, TDelegate handler) 
    { 
     if (add == null) { throw new ArgumentNullException("add"); } 
     if (remove == null) { throw new ArgumentNullException("remove"); } 
     if (handler == null) { throw new ArgumentNullException("handler"); } 
     add(handler); 
     return this.Add(() => remove(handler)); 
    } 

    public void Clear() 
    { 
     lock (_disposables) 
     { 
      var disposables = _disposables.ToArray(); 
      _disposables.Clear(); 
      Array.ForEach(disposables, d => d.Dispose()); 
     } 
    } 

    public void Dispose() 
    { 
     lock (_disposables) 
     { 
      if (!_disposed) 
      { 
       this.Clear(); 
      } 
      _disposed = true; 
     } 
    } 

    public IEnumerator<IDisposable> GetEnumerator() 
    { 
     lock (_disposables) 
     { 
      return _disposables.ToArray().AsEnumerable().GetEnumerator(); 
     } 
    } 

    IEnumerator IEnumerable.GetEnumerator() 
    { 
     return this.GetEnumerator(); 
    } 

    public bool IsDisposed 
    { 
     get 
     { 
      return _disposed; 
     } 
    } 
}