如何清除重播主题上的缓冲区

本文关键字：缓冲区 何清除 清除 | 更新日期: 2024-11-07 21:17:17

如何清除ReplaySubject上的缓冲区？

我需要定期清除缓冲区（在我的情况下作为一天结束事件），以防止ReplaySubject不断增长并最终吞噬所有内存。

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

如何清除重播主题上的缓冲区

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() 。如果需要，您可以轻松添加同步锁。

它的工作原理是将一系列重播主题嵌套在主重播主题中。外部重播主体（_subjects）包含一个内部重播主体（_currentSubject）的缓冲区，并在构造时填充。

OnXXX方法调用_currentSubject重播主题。

观察者订阅嵌套重播主题的串联投影（以 _concatenatedSubjects 形式保存）。由于 _subjects 的缓冲区大小仅为 1，因此新订阅者仅获取最近ReplaySubject的事件。

每当我们需要"清除缓冲区"时，都会OnCompleted现有的_currentSubject，并将新的重播主题添加到_subjects并成为新的_currentSubject。

增强

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

您可能已经有一个可观察的数据源，在这种情况下，这是另一种解决方案。这个使用现有 RX 结构的组合，而不是构建我个人警惕的自己的 ISubject。

public class ClearableReplaySubject<TSource, TClearTrigger> : IConnectableObservable<TSource>
{
    private readonly IConnectableObservable<IObservable<TSource>> _underlying;
    private readonly SerialDisposable _replayConnectDisposable = new SerialDisposable();
    public ClearableReplaySubject(IObservable<TSource> src, IObservable<TClearTrigger> clearTrigger)
    {
        _underlying = clearTrigger.Select(_ => Unit.Default).StartWith(Unit.Default)
            .Select(_ =>
            {
                var underlyingReplay = src.Replay();
                _replayConnectDisposable.Disposable = underlyingReplay.Connect();
                return underlyingReplay;
            })
            .Replay(1);
    }
    public IDisposable Subscribe(IObserver<TSource> observer)
    {
        return _underlying.Switch().Subscribe(observer);
    }
    public IDisposable Connect()
    {
        return new CompositeDisposable(_underlying.Connect(), _replayConnectDisposable.Disposable);
    }
}

如果将以下扩展方法添加到 ObservableEx：

public static class ObservableEx
{
    public static IConnectableObservable<TItem> ReplayWithReset<TItem, TReset>(this IObservable<TItem> src, IObservable<TReset> resetTrigger)
    {
        return new ClearableReplaySubject<TItem, TReset>(src, resetTrigger);
    }
}

然后，您可以获取源并添加.重播与重置（...）与重置触发器可观察。这可能是一个计时器或其他什么。

var replay = sourceObservable.ReplayWithReset(triggerObservable);
var connection = replay.Connect();

连接的行为方式与重播相同。

好吧，

我不知道c#，但我设法在重播主题rxdart中完成了它。 至于重播主题，它使用队列来缓存事件，所以我 修改了重播主题类。

  1. 我更改了所有队列 到列表
  2. 添加了 onRemove 方法，该方法将从中删除事件 查克的名单。

原版重播主题 ：

class ReplaySubject<T> extends Subject<T> implements ReplayStream<T> {
final Queue<T> _queue;
final int _maxSize;
/// Constructs a [ReplaySubject], optionally pass handlers for
/// [onListen], [onCancel] and a flag to handle events [sync].
///
/// See also [StreamController.broadcast]
factory ReplaySubject({
    int maxSize,
    void Function() onListen,
    void Function() onCancel,
    bool sync = false,
}) {
    // ignore: close_sinks
    final controller = StreamController<T>.broadcast(
        onListen: onListen,
        onCancel: onCancel,
        sync: sync,
    );
    final queue = Queue<T>();
    return ReplaySubject<T>._(
        controller,
        Rx.defer<T>(
                () => controller.stream.startWithMany(queue.toList(growable: false)),
            reusable: true,
        ),
        queue,
        maxSize,
    );
}
ReplaySubject._(
    StreamController<T> controller,
    Stream<T> stream,
    this._queue,
    this._maxSize,
    ) : super(controller, stream);
@override
void onAdd(T event) {
    if (_queue.length == _maxSize) {
        _queue.removeFirst();
    }
    _queue.add(event);
}
@override
List<T> get values => _queue.toList(growable: false);
}

修改重播主题：

class ModifiedReplaySubject<T> extends Subject<T> implements ReplayStream<T> {
final List<T> _list;
final int _maxSize;
/// Constructs a [ModifiedReplaySubject], optionally pass handlers for
/// [onListen], [onCancel] and a flag to handle events [sync].
///
/// See also [StreamController.broadcast]
factory ModifiedReplaySubject({
    int maxSize,
    void Function() onListen,
    void Function() onCancel,
    bool sync = false,
}) {
    // ignore: close_sinks
    final controller = StreamController<T>.broadcast(
        onListen: onListen,
        onCancel: onCancel,
        sync: sync,
    );
    final queue = List<T>();
    return ModifiedReplaySubject<T>._(
        controller,
        Rx.defer<T>(
                () => controller.stream.startWithMany(queue.toList(growable: false)),
            reusable: true,
        ),
        queue,
        maxSize,
    );
}
ModifiedReplaySubject._(
    StreamController<T> controller,
    Stream<T> stream,
    this._list,
    this._maxSize,
    ) : super(controller, stream);
@override
void onAdd(T event) {
    if (_list.length == _maxSize) {
        _list.removeAt(0);
    }
    _list.add(event);
}
void onRemove(T event) {
    _list.remove(event);
}
@override
List<T> get values => _list.toList(growable: false);
}
相关文章：