c# - Rx.NET 中是否有一个 Subject 实现在功能上类似于 BehaviorSubject 但仅在值发生变化时才会发出？

Question

有没有 Subject Rx.NET 中的实现在功能上类似于 BehaviorSubject 但只有在它发生变化时才会发出下一个值？

我对 Reactive Extensions 比较陌生，我似乎找不到类似的东西，尽管这种模式感觉像是 INotifyPropertyChanged 的自然替代品。 .

我天真的实现是封装BehaviorSubject<T>像下面。与使用 Observable.DistinctUntilChanged 创建可组合的可观察对象相比，这有什么缺点吗？ ？

    public class DistinctSubject<T> : SubjectBase<T>
    {
        private readonly BehaviorSubject<T> _subject;

        public DistinctSubject(T initialValue) =>
            _subject = new BehaviorSubject<T>(initialValue);

        public T Value 
        { 
            get => _subject.Value;
            set => this.OnNext(value);
        }

        public override bool HasObservers => _subject.HasObservers;

        public override bool IsDisposed => _subject.IsDisposed;

        public override void Dispose() => _subject.Dispose(); 

        public override void OnCompleted() => _subject.OnCompleted();   

        public override void OnError(Exception error) => _subject.OnError(error);

        public override void OnNext(T value)
        {
            if (!EqualityComparer<T>.Default.Equals(value, _subject.Value))
            {
                _subject.OnNext(value);
            }
        }

        public override IDisposable Subscribe(IObserver<T> observer) =>
            _subject.Subscribe(observer);
    }

Answer 1

浏览了一下 source code 之后的 BehaviorSubject<T>类，看来你的 DistinctSubject<T>在 OnError 的情况下，实现的行为会有所不同。后面跟着一个 OnNext :

var subject = new DistinctSubject<int>(2021);
subject.OnError(new ApplicationException());
subject.OnNext(2022); // throws ApplicationException

这将抛出，同时对 BehaviorSubject<T> 做同样的事情不会抛出(OnNext 被忽略)。

我的建议是使用 DistinctUntilChanged运算符在实现中，像这样:

public class DistinctSubject<T> : ISubject<T>, IDisposable
{
    private readonly BehaviorSubject<T> _subject;
    private readonly IObservable<T> _distinctUntilChanged;

    public DistinctSubject(T initialValue, IEqualityComparer<T> comparer = default)
    {
        _subject = new BehaviorSubject<T>(initialValue);
        _distinctUntilChanged = _subject.DistinctUntilChanged(
            comparer ?? EqualityComparer<T>.Default);
    }

    public T Value => _subject.Value;
    public void OnNext(T value) => _subject.OnNext(value);
    public void OnError(Exception error) => _subject.OnError(error);
    public void OnCompleted() => _subject.OnCompleted();

    public IDisposable Subscribe(IObserver<T> observer) =>
        _distinctUntilChanged.Subscribe(observer);

    public void Dispose() => _subject.Dispose();
}

如果您担心对象的不必要分配，那么您还没有熟悉 Rx 的精神。这个库是关于功能和易用性的，而不是关于 performance或效率!