Rx中的觀察者數量主題

Question

使用Rx，獲取主題中當前觀察者數量的最佳方法是什麼？

我有，我要發佈消息的情況下，但前提是有觀察員。如果沒有觀察員，我需要做別的事情。

爲了解決這個問題，我所做的就是創建了自己的ISubject實現，公開內部IObserver收集的計數。我確信必須有一個開箱即用的方式，我只是不完全熟悉Rx所提供的。

謝謝！

Answer 1

你應該避免實現你自己觀察到的（或學科）實現只要有可能。

你當然可以嘗試編寫一個包裝類來提供幫助。

試試這個：

public class Countable 
{ 
    private int _count; 
    public int Count { get { return _count; } } 
    public IObservable<T> GetCountable<T>(IObservable<T> source) 
    { 
     return Observable.Create<T>(o => 
     { 
      Interlocked.Increment(ref _count); 
      var subscription = source.Subscribe(o); 
      var decrement = Disposable.Create(() => 
      { 
       Interlocked.Decrement(ref _count); 
      }); 
      return new CompositeDisposable(subscription, decrement); 
     }); 
    } 
} 

然後你可以這樣寫代碼：

var xs = new Subject<int>(); 
var countable = new Countable(); 
var ys = countable.GetCountable(xs); 
Console.WriteLine(countable.Count); 
var s1 = ys.Subscribe(y => { }); 
Console.WriteLine(countable.Count); 
var s2 = ys.Subscribe(y => { }); 
Console.WriteLine(countable.Count); 
s1.Dispose(); 
Console.WriteLine(countable.Count); 
s2.Dispose(); 
Console.WriteLine(countable.Count); 

我運行這個結果是：

0 
1 
2 
1 
0 

Answer 2

使用Subject<T>.HasObservers財產。

Source Code

我完全不記得它推出的時候，但我敢肯定，這是不是永遠存在的。它可能是在Rx 2.0中添加的。

Answer 3

使用subject.observers.length，例如：

import {Subject} from 'rxjs' 

let subject = new Subject() 
let s1 = subject.subscribe(v => console.log('observerA: ' + v)) 

subject.next(1) // observerA: 1 
console.log(subject.observers.length) // 1 


let s2 = subject.subscribe(v => { 
    console.log('observerB: ' + v) 
    if(v===3) s2.unsubscribe() 
}) 

subject.next(2) // observerA: 2 
console.log(subject.observers.length) // 2 

subject.next(3) // observerA: 3 
console.log(subject.observers.length) // 1