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