如何清除ReplaySubject上的缓冲区? [英] How can I clear the buffer on a ReplaySubject?

问题描述

如何清除 ReplaySubject 上的缓冲区?

我定期需要清除缓冲区(对于我来说，这是一天结束的事件)，以防止 ReplaySubject 持续增长并最终耗尽所有内存.

Periodically I need to clear the buffer (as an end of day event in my case) to prevent the ReplaySubject continually growing and eventually eating all the memory.

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

Ideally I want to keep the same ReplaySubject as the client subscriptions are still good.

推荐答案

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

ReplaySubject doesn't offer a means to clear the buffer, but there are several overloads to constrain its buffers in different ways:

• 保留项目的最大 TimeSpan
• 最大项目数
• 上述内容的组合，一旦满足任一条件，该物品就会掉落.

这是一个非常有趣的问题-我决定看看使用现有主题和运算符(如这些功能非常强大).事实证明，这相当简单.

This was quite an interesting problem - I decided to see how easy it would be to implement a variation of ReplaySubject you can clear - using existing subjects and operators (as these are quite robust). Turns out it was reasonably straightforward.

我已经通过内存探查器运行它来检查它是否做对了.调用 Clear()刷新缓冲区，否则它就像常规的无界 ReplaySubject :

I've run this through a memory profiler to check it does the right thing. Call Clear() to flush the buffer, otherwise it works just like a regular unbounded 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().如果需要，您可以轻松添加同步锁.

Respect usual rules (as with any Subject) and don't call methods on this class concurrently - including Clear(). You could add synchronization locks trivially if needed.

它通过在主ReplaySubject内嵌套一系列ReplaySubject来工作.外部ReplaySubject( _subjects )容纳一个正好是内部ReplaySubject( _currentSubject )的缓冲区，并且在构造时填充该缓冲区.

It works by nesting a sequence of ReplaySubjects inside a master ReplaySubject. The outer ReplaySubject (_subjects) holds a buffer of exactly one inner ReplaySubject (_currentSubject), and it is populated on construction.

OnXXX 方法调用 _currentSubject ReplaySubject.

The OnXXX methods call through to the _currentSubject ReplaySubject.

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

Observers are subscribed to a concatenated projection of the nested ReplaySubjects (held in _concatenatedSubjects). Because the buffer size of _subjects is just 1, new subscribers acquire the events of only the most recent ReplaySubject onwards.

每当我们需要清除缓冲区"时，现有的 _currentSubject 为 OnCompleted ，并且将新的ReplaySubject添加到 _subjects 并变为新的 _currentSubject .

Whenever we need to "clear the buffer", the existing _currentSubject is OnCompleted and a new ReplaySubject is added to _subjects and becomes the new _currentSubject.

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

Following @Brandon's suggestion, I created a version of RollingReplaySubject that uses either a TimeSpan or an input stream to signal buffer clearing. I created a Gist for this here: https://gist.github.com/james-world/c46f09f32e2d4f338b07

这篇关于如何清除ReplaySubject上的缓冲区?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！