c# - 如何清除 ReplaySubject 上的缓冲区？

Question

如何清除 ReplaySubject 上的缓冲区？

我需要定期清除缓冲区(在我的例子中作为一天结束的事件)以防止 ReplaySubject 不断增长并最终吃掉所有内存。

理想情况下，我希望保持相同的 ReplaySubject，因为客户端订阅仍然很好。

Answer 1

ReplaySubject 不提供清除缓冲区的方法，但有几种重载以不同方式限制其缓冲区:

• 项目保留的最大TimeSpan
• 最大项目数
• 上述的组合，只要满足其中一个条件就会掉落元素。

一个可清除的 ReplaySubject

这是一个非常有趣的问题 - 我决定看看使用现有的主题和运算符(如这些非常坚固)。结果证明它相当简单。

我已经通过内存分析器运行它来检查它是否正确。调用 Clear() 来刷新缓冲区，否则它就像一个普通的无界 ReplaySubject 一样工作:

public class RollingReplaySubject<T> : ISubject<T>
{
    private readonly ReplaySubject<IObservable<T>> _subjects;
    private readonly IObservable<T> _concatenatedSubjects;
    private ISubject<T> _currentSubject;

    public RollingReplaySubject()
    {
        _subjects = new ReplaySubject<IObservable<T>>(1);
        _concatenatedSubjects = _subjects.Concat();
        _currentSubject = new ReplaySubject<T>();
        _subjects.OnNext(_currentSubject);
    }

    public void Clear()
    {
        _currentSubject.OnCompleted();
        _currentSubject = new ReplaySubject<T>();
        _subjects.OnNext(_currentSubject);
    }

    public void OnNext(T value)
    {
        _currentSubject.OnNext(value);
    }

    public void OnError(Exception error)
    {
        _currentSubject.OnError(error);
    }

    public void OnCompleted()
    {
        _currentSubject.OnCompleted();
        _subjects.OnCompleted();     
        // a quick way to make the current ReplaySubject unreachable
        // except to in-flight observers, and not hold up collection
        _currentSubject = new Subject<T>();       
    }

    public IDisposable Subscribe(IObserver<T> observer)
    {
        return _concatenatedSubjects.Subscribe(observer);
    }
}

遵守通常的规则(与任何 Subject 一样)并且不要同时调用此类上的方法 - 包括 Clear()。如果需要，您可以简单地添加同步锁。

它通过在主 ReplaySubject 中嵌套一系列 ReplaySubjects 来工作。外层 ReplaySubject (_subjects) 恰好包含一个内部 ReplaySubject (_currentSubject) 的缓冲区，并在构造时填充。

OnXXX 方法调用 _currentSubject ReplaySubject。

观察者订阅嵌套 ReplaySubjects 的串联投影(保存在 _concatenatedSubjects 中)。因为 _subjects 的缓冲区大小仅为 1，所以新订阅者只会获取最近的 ReplaySubject 之后的事件。

每当我们需要“清除缓冲区”时，现有的 _currentSubject 是 OnCompleted 并且一个新的 ReplaySubject 被添加到 _subjects 并成为新的 _currentSubject。

增强功能

按照@Brandon 的建议，我创建了一个 RollingReplaySubject 版本，它使用 TimeSpan 或输入流来指示缓冲区清除。我在这里为此创建了一个要点:https://gist.github.com/james-world/c46f09f32e2d4f338b07