Reactive的“缓冲直到安静”行为？

Question

我的问题有点像 Nagle algorithm 旨在解决的问题，但不完全一样。我想要将来自IObservable<T> 的OnNext 通知缓冲到IObservable<IList<T>>s 序列中，如下所示：

1. 当第一个T 通知到达时，将其添加到缓冲区并开始倒计时
2. 如果另一个T 通知在倒计时到期之前到达，请将其添加到缓冲区并重新开始倒计时
3. 一旦倒计时结束（即生产者已经静默一段时间），将所有缓冲的 T 通知作为单个聚合 IList<T> 通知转发。
4. 如果在倒计时结束前缓冲区大小超过某个最大值，请无论如何发送。

IObservable<IList<T>> Buffer(this IObservable<T>, Timespan, int, IScheduler) 看起来很有希望，但它似乎会定期发送汇总通知，而不是“在第一个通知到达时启动计时器并在其他通知到达时重新启动它”我想要的行为，而且它也如果没有从下方生成通知，则在每个时间窗口结束时发送一个空列表。

我确实不想要删除任何T 通知；只是缓冲它们。

这样的东西存在吗，还是我需要自己写？

Answer 1

SO 上存在一些类似的问题，但不完全是这样。 这是一个可以解决问题的扩展方法。

public static IObservable<IList<TSource>> BufferWithThrottle<TSource>
                                          (this IObservable<TSource> source,
                                           int maxAmount, TimeSpan threshold)
{
    return Observable.Create<IList<TSource>>((obs) =>
    {
        return source.GroupByUntil(_ => true,
                                   g => g.Throttle(threshold).Select(_ => Unit.Default)
                                         .Merge( g.Buffer(maxAmount).Select(_ => Unit.Default)))
                     .SelectMany(i => i.ToList())
                     .Subscribe(obs);
    });
}

Answer 2

有趣的运算符。 Supertopi 的答案是一个很好的答案，但是可以进行改进。如果maxAmount 很大，并且/或者通知率很高，那么使用Buffer 将通过分配很快被丢弃的缓冲区来烧毁GC。

为了在达到maxAmount 后关闭每个GroupBy Observable，您无需捕获所有这些元素的Buffer 即可知道它何时已满。根据 Supertopi 的回答，您可以将其稍微更改为以下内容。它不会收集maxAmount 元素中的Buffer，而是在看到流中的maxAmount 元素后发出信号。

public static IObservable<IList<TSource>> BufferWithThrottle<TSource>(this IObservable<TSource> source, int maxAmount, TimeSpan threshold)
{
    return Observable.Create<IList<TSource>>((obs) =>
    {
        return source.GroupByUntil(_ => true,
                                   g => g.Throttle(threshold).Select(_ => Unit.Default)
                                         .Merge(g.Take(maxAmount)
                                                 .LastAsync()
                                                 .Select(_ => Unit.Default)))
                     .SelectMany(i => i.ToList())
                     .Subscribe(obs);
    });
}

Answer 3

不错的解决方案。在我看来，使用现有运算符创建行为只是为了方便而不是为了性能。

此外，我们应该始终返回 IEnumerable 而不是 IList。 返回最小派生类型 (IEnumerable) 将为您留出最大的余地来更改底层实现。

这是我实现自定义运算符的版本。

public static IObservable<IEnumerable<TValue>> BufferWithThrottle<TValue>(this IObservable<TValue> @this, int maxAmount, TimeSpan threshold)
    {
        var buffer = new List<TValue>();

        return Observable.Create<IEnumerable<TValue>>(observer =>
        {
            var aTimer = new Timer();
            void Clear()
            {
                aTimer.Stop();
                buffer.Clear();
            }
            void OnNext()
            {
                observer.OnNext(buffer);
                Clear();
            }
            aTimer.Interval = threshold.TotalMilliseconds;
            aTimer.Enabled = true;
            aTimer.Elapsed += (sender, args) => OnNext();
            var subscription = @this.Subscribe(value =>
            {
                buffer.Add(value);
                if (buffer.Count >= maxAmount)
                    OnNext();
                else
                {
                    aTimer.Stop();
                    aTimer.Start();
                }
            });
            return Disposable.Create(() =>
            {
                Clear();
                subscription.Dispose();
            });
        });
    }

通过对比其他解决方案的性能测试，它可以节省高达 30% 的 CPU 功率并解决内存问题。