system.reactive - “rate limit”消耗Observable的最佳方法是什么？

Question

我有很多事件要发生，我必须无损失地执行所有事件，但是我想确保在适当的时间段对它们进行缓冲和使用。有人有解决办法吗？

我在Rx中找不到任何可以在不丢失事件的情况下做到这一点的操作符( throttle -松开事件)。我还考虑了缓冲，延迟等。找不到很好的解决方案。

我试图将计时器放在中间，但是不知怎么不起作用:

GetInitSequence()
            .IntervalThrottle(TimeSpan.FromSeconds(5))
            .Subscribe(
                item =>
                    {
                        Console.WriteLine(DateTime.Now);
                        // Process item
                    }
            );

public static IObservable<T> IntervalThrottle<T>(this IObservable<T> source, TimeSpan dueTime)
    {
        return Observable.Create<T>(o =>
            {
                return source.Subscribe(x =>
                    {
                        new Timer(state => 
                            o.OnNext((T)state), x, dueTime, TimeSpan.FromMilliseconds(-1));
                    }, o.OnError, o.OnCompleted);
        });
    }

Answer 1

这个问题不是100％清楚的，所以我要作一些假设。
Observable.Delay不是您想要的，因为这将延迟每个事件到达的时间，而不是创建均匀的时间间隔进行处理。
Observable.Buffer不是您想要的，因为这将导致将每个给定间隔中的所有事件传递给您，而不是一次传递给您。

因此，我相信您正在寻找一种解决方案，该解决方案可以创建一种节拍器，该节拍器会滴答作响，并为您提供每个滴答声的事件。这可以是天真的，它是用节拍器的Observable.Interval和将其连接到源的Zip构造的:

var source = GetInitSequence();
var trigger = Observable.Interval(TimeSpan.FromSeconds(5));    
var triggeredSource = source.Zip(trigger, (s,_) => s); 
triggeredSource.Subscribe(item => Console.WriteLine(DateTime.Now));

这将每5秒触发一次(在上面的示例中)，并依次为您提供原始项目。

此解决方案的唯一问题是，如果您在10秒钟内没有其他源元素(例如)，则在源元素到达时，由于某些“触发”事件正坐在那里等待它们，因此它们将被立即发送出去。 。该场景的大理石图:

source:  -a-b-c----------------------d-e-f-g
trigger: ----o----o----o----o----o----o----o
result:  ----a----b----c-------------d-e-f-g

这是一个非常合理的问题。这里已经有两个问题可以解决:

Rx IObservable buffering to smooth out bursts of events

A way to push buffered events in even intervals

提供的解决方案是主要的 Drain扩展方法和辅助的 Buffered扩展。我已经将它们修改为简单得多(不需要 Drain，只需使用 Concat)。用法是:

var bufferedSource = source.StepInterval(TimeSpan.FromSeconds(5));

扩展方法 StepInterval:

public static IObservable<T> StepInterval<T>(this IObservable<T> source, TimeSpan minDelay)
{
    return source.Select(x => 
        Observable.Empty<T>()
            .Delay(minDelay)
            .StartWith(x)
    ).Concat();
}