这可观察到的是热的吗

本文关键字：观察 | 更新日期: 2023-09-27 17:58:17

这不会让可观察到的东西变热吗？

using System;
using System.Reactive;
using System.Reactive.Disposables;
using System.Reactive.Linq;
namespace ObservableNumberGenerator.ObservableImplementationReliesOnOperator.Hot
{
    public class RandomNumbers : IObservable<int>, IDisposable
    {
        protected Random _random = null;
        protected int _maxNumbersToGenerate;
        protected int _startAfterMilliseconds = 1000;
        protected int _generateEveryMilliseconds = 1000;
        protected IObservable<int> _innerObservable = null;
        protected IDisposable _innerSubscription = null;
        protected bool _completed = false;
        private bool disposedValue = false;

        public RandomNumbers(int maxNumbersToGenerate,
            int startAfterMilliseconds = 1000, 
            int generateEveryMilliseconds = 1000)
        {
            _maxNumbersToGenerate = maxNumbersToGenerate;
            _startAfterMilliseconds = startAfterMilliseconds;
            _generateEveryMilliseconds = generateEveryMilliseconds;
            _random = new Random();
            _innerObservable = Observable.Timer(
                TimeSpan.FromMilliseconds(_startAfterMilliseconds),
                TimeSpan.FromMilliseconds(_generateEveryMilliseconds))
                .Select(v => GenerateNumber())
                .Take(_maxNumbersToGenerate);
            _innerSubscription = _innerObservable.Subscribe(OnNext, OnError, OnCompleted);
        }
        protected virtual void OnCompleted()
        {
            _completed = true;
        }
        protected virtual void OnError(Exception ex)
        {
        }
        protected virtual void OnNext(int value)
        {
        }
        protected virtual int GenerateNumber()
        {
            return _random.Next();
        }
        public IDisposable Subscribe(IObserver<int> observer)
        {
            if (observer == null) throw new ArgumentNullException("observer");
            if (_completed)
            {
                observer.OnCompleted();
                return Disposable.Empty;
            }
            else
            {
                return _innerObservable.Subscribe(observer);
            }
        }
        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    _innerSubscription.Dispose();
                }
                disposedValue = true;
            }
        }
        public void Dispose()
        {
            Dispose(true);
        }
    }
}

或者我会让到Publish和Connect来让它变热吗？

public class RandomNumbers : IObservable<int>, IDisposable
{
    protected IConnectableObservable<int> _innerObservable = null;        
    public RandomNumbers(int maxNumbersToGenerate,
        int startAfterMilliseconds = 1000, 
        int generateEveryMilliseconds = 1000)
    {
        _maxNumbersToGenerate = maxNumbersToGenerate;
        _startAfterMilliseconds = startAfterMilliseconds;
        _generateEveryMilliseconds = generateEveryMilliseconds;
        _random = new Random();
        _innerObservable = Observable
            .Timer(TimeSpan.FromMilliseconds(_startAfterMilliseconds),
            TimeSpan.FromMilliseconds(_generateEveryMilliseconds))
            .Select(v => GenerateNumber())
            .Take(_maxNumbersToGenerate)
            .Publish();
        _innerObservable.Connect();
        _innerSubscription = _innerObservable.Subscribe(OnNext, OnError, OnCompleted);
    }
    ...
}

这可观察到的是热的吗

您的第一个代码是冷可观察的。以下是一些简单的客户端代码：

    static void Main(string[] args)
    {
        var x = new RandomNumbers(10, 0, 500);
        x.Timestamp().Subscribe(i => Console.WriteLine("First sub: {0} {1}", i.Timestamp.DateTime.ToString("O"), i.Value));
        Thread.Sleep(1000);
        x.Timestamp().Subscribe(i => Console.WriteLine("Second sub: {0} {1}", i.Timestamp.DateTime.ToString("O"), i.Value));
        Console.Read();
    }

通过冷观测，您将看到"第一个子"行与"第二个子"行不匹配。对于热实现，它们确实如此。你还会注意到，在冷实现中，有相同数量的"第一子"answers"第二子"行。通过热实现，可以在全局范围内观察到最大值10。

只是为了澄清热与冷。假设有一个静态/稳定的数据流，您会得到这个大理石图，显示在t0和t1:订阅时订阅的差异

hot source:  A---B---C---D---
sub at t0 :  A---B---C---D---
sub at t1 :      B---C---D---
cold source: A---B---C---D---
sub at t0  : A---B---C---D---
sub at t1  :     A---B---C---

然而，在我们的案例中，我们有一个随机流：

hot source:  R()-R()-R()-R()-
sub at t0 :  R1--R2--R3--R4--
sub at t1 :      R2--R3--R4--
cold source: R()-R()-R()-R()-
sub at t0  : R1--R2--R3--R4--
sub at t1  :     R5--R6--R7--

Random()调用的结果在热的可观察到的情况下共享，而在冷的情况下则不共享。如果按索引进行测量，则热可观察订阅会得到不同的值，但它们在给定的时间t会收到相同的值。在稳定的源中，冷可观察订阅按索引匹配，但不按时间匹配。在一个不稳定的源中，就像我们的情况一样，冷的可观测值在索引或时间上都不匹配。

正如Enigmatity所指出的，正如Shlomo的回答似乎也暗示的那样，仅仅因为你在构造函数中订阅了可观察对象，就不会让它"共享"，从而想要让它变得热门。

我仔细思考了为什么应该这样，然后我查看了Timer运算符上Subscribe方法的源代码，意识到了我之前知道的一些事情，但由于我看得太仔细而忘记了。

大多数运算符都是这样实现的：每次在Timer运算符上调用Subscribe方法时，都会创建一个内部类的新实例，该实例在默认池调度程序上启动一个新序列。

所以，在Timer类的某个地方写着：

namespace System.Reactive.Linq.ObservableImpl
{
    internal class Timer : Producer<long>
    {
        protected override IDisposable Run(
                               IObserver<long> observer, 
                               IDisposable cancel, 
                               Action<IDisposable> setSink)
        {
            if (this._period.HasValue)
            {
                Timer.TimerImpl timerImpl = 
                    new Timer.TimerImpl(this, observer, cancel);
                setSink(timerImpl);
                return timerImpl.Run();
            }
            ...
        }
    }
}

再加上这个事实，在我的自定义observable的Subscribe方法中，我这样做：

public IDisposable Subscribe(IObserver<int> observer)
{
    if (_completed)
    {
        ...
    }
    else
    {
        return _innerObservable.Subscribe(observer);
    }
}

如果我通过内部观察器引导所有观测，我可以在不使用Publish和Connect或RefCount运算符的情况下使这种可观测到的热态：

using System;
using System.Collections.Generic;
using System.Reactive.Disposables;
using System.Reactive.Linq;
namespace ObservableNumberGenerator.ObservableImplementationReliesOnOperator.Hot
{
    public class RandomNumbers : IObservable<int>, IDisposable
    {
        protected Random _random = null;
        protected int _maxNumbersToGenerate;
        protected int _startAfterMilliseconds = 1000;
        protected int _generateEveryMilliseconds = 1000;
        protected List<IObserver<int>> _observers = 
                            new List<IObserver<int>>();
        protected IObservable<int> _innerObservable = null;
        protected IDisposable _innerSubscription = null;
        protected bool _completed = false;
        private bool disposedValue = false;
        public RandomNumbers(int maxNumbersToGenerate,
            int startAfterMilliseconds = 1000, int generateEveryMilliseconds = 1000)
        {
            _maxNumbersToGenerate = maxNumbersToGenerate;
            _startAfterMilliseconds = startAfterMilliseconds;
            _generateEveryMilliseconds = generateEveryMilliseconds;
            _random = new Random();
            _innerObservable = Observable
                .Timer(
                  TimeSpan.FromMilliseconds(_startAfterMilliseconds),
                  TimeSpan.FromMilliseconds(_generateEveryMilliseconds))
                .Select(v => GenerateNumber())
                .Take(_maxNumbersToGenerate);
            _innerSubscription = 
                _innerObservable.Subscribe(OnNext, OnError, OnCompleted);
        }
        protected virtual void OnCompleted()
        {
            _completed = true;
            foreach (var observer in _observers)
                observer.OnCompleted();
        }
        protected virtual void OnError(Exception ex)
        {
            _completed = true;
            foreach (var observer in _observers)
                observer.OnError(ex);
        }
        protected virtual void OnNext(int value)
        {
            foreach (var observer in _observers)
                observer.OnNext(value);
        }
        protected virtual int GenerateNumber()
        {
            return _random.Next();
        }
        public IDisposable Subscribe(IObserver<int> observer)
        {
            if (observer == null) throw new ArgumentNullException("observer");
            _observers.Add(observer);
            if (_completed)
            {
                observer.OnCompleted();
            }
            return Disposable.Empty;
        }
        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    _innerSubscription.Dispose();
                }
                disposedValue = true;
            }
        }
        public void Dispose()
        {
            Dispose(true);
        }
    }
}

任何想这样做的人都要注意：

我所有的例子都只是练习，目的是让自己了解Rx的内部运作。

正如Enigmativity正确指出的那样，在生产代码中不能以这种方式实现任何可观察的东西。

首先，我的List<IObserver<T>>和_innerObserver中的观察员都不是安全的观察员。在Rx源中，他们将我们传递的每个观测器转换为一个安全观测器，该观测器在每个OnNext、OnError和OnCompleted实现中都有try/finally子句，以防止出现异常。有关详细信息，请查看System.Reactive.SafeObserver<TSource>类和System.Reactive.AnonymousObserver<T>类的源代码。

相关文章：