有没有告诉订阅观察者数量的 Rx.Net Operator？答案

【问题标题】：Is there a Rx.Net Operator that tells the number of subscribed observers?有没有告诉订阅观察者数量的 Rx.Net Operator？
【发布时间】：2020-02-12 09:43:37
【问题描述】：

如果我正确地将观察者部署到长时间运行的服务（暴露 IObservable 的服务），我正在尝试调试。我想知道是否有一个操作员或我们可以创建的东西来记录活跃观察者的数量，比如。

public class NewsService
{
    IObservable<Article> GetArticles();
} 


NewsService.Instance
    .GetArticles()
    .DoCount(x=> Trace.Writeline("The current count is {x}"))
    .Subscribe();

提出了一个解决方案here，它适用于主题。如果我们无权访问主题并且库正在公开IObservable，该怎么办。

【问题讨论】：

标签： system.reactive

【解决方案1】：

一般来说，没有定义任意可观察序列的订阅者计数的概念。

对于像Observable.Interval 这样的冷可观察对象，每次订阅可观察对象时，都会创建一个新的管道实例，从它的角度来看，它一次只能看到一个观察者。

尽管如此，我们可以预热一个寒冷的 observable，然后观察订阅的来来去去。

    public static IObservable<T> RefCount<T>(this IObservable<T> source, Action<int> onChange)
    {
        var subscribers = 0;
        var shared = source.Publish().RefCount();
        void callback(int count) => onChange(Interlocked.Add(ref subscribers, count));

        return Observable.Create<T>(observer =>
        {
            callback(+1);
            var subscription = shared.Subscribe(observer);
            var dispose = Disposable.Create(() => callback(-1));

            return new CompositeDisposable(subscription, dispose);
        });
    }

演示

        var values = 
            Observable
            .Interval(TimeSpan.FromSeconds(0.1))
            .RefCount(count => Console.WriteLine($"Subscribers: {count}"));

        values.Take(5).Subscribe();
        values.Take(10).Subscribe();
        values.Take(15).Subscribe();

输出

Subscribers: 1
Subscribers: 2
Subscribers: 3
Subscribers: 2
Subscribers: 1
Subscribers: 0

现在，这行得通，因为我们拥有父 observable 的共享视图。所以尽量让所有订阅都指向同一个实例。

_articles = GetArticles().RefCount(count => Console.WriteLine($"Subscribers: {count}")));
...
_articles.Subscribe();

【讨论】：