'滑动' RX.net .Throttle() 窗口 [重复]答案

【问题标题】：'Sliding' RX.net .Throttle() Window [duplicate]'滑动' RX.net .Throttle() 窗口 [重复]
【发布时间】：2018-03-03 06:45:57
【问题描述】：

我对一个超出正常限制的可观察流有某种特殊需求/要求，我不完全确定如何做到这一点：

基本上，我有一个可观察的流，最初来自正常事件，如下所示：

var someEventObservable = Observable.FromEventPattern<SomeEventHandler, SomeEventArgs>(
    handler => this.ColumnWidthChanged += handler,
    handler => this.ColumnWidthChanged -= handler)
    .Select(_ => Unit.Default);

现在由于这些事件可以快速连续发生，我只需要知道它是否在给定时间范围内至少发生一次，我通常会使用 .Throttle()，即像这样：

var someThrottledEventObservable = someEventObservable
    .Throttle(TimeSpan.FromMilliseconds(300));

但我的实际要求更进一步：如果在该限制 TimeSpan / dueTime 内引发了一个事件，并且如果在第一个事件之后引发另一个事件但仍在该 dueTime 内，我希望限制流从 0 重新开始再次等待时间并再等待 300 毫秒……如果引发了另一个事件，请再次重新开始/延长该时间……依此类推。只有在原始的或重新启动的 TimeSpan(s)/dueTime 内没有引发其他事件时，someThrottledEventObservable 才会产生一个新的 Unit 实例。

我希望这是有道理的 - 但基本上我想要/需要一个受限制的事件流，只要源流在给定时间内停止产生新事件并且如果在该等待时间内发生新事件，受限制的流应该产生一个事件重新开始等待。

或者：在持续的事件“风暴”中，仅 .Throttle() 就会每 300 毫秒产生一个新单元（在上面的示例中），但是每当一个或多个事件被触发但没有新事件发生时，我都想要一个新单元之后的 300 毫秒冷却期内。

我该怎么做？

【问题讨论】：

不熟悉 rx.net 语义，但听起来 Throttle 的行为类似于 sample 而不是 debounce，reactivex.io 声称这是正在运行的操作。 Debounce 应该像您描述的那样工作（基于文档），并且您的问题的一个公平的改写可能是“我如何在 Reactive Extensions for .NET 中去抖动”。
我认为 Throttle 已经在做你所期望的。如果在指定时间段后没有其他事件，它只会发出最后一个事件。
@nikoniko 是的-我一定是在这里做错了..不知道为什么，但是在重新阅读文档后，它清楚地表明它应该/确实按照我的意愿行事.. 显然有些事情否则会导致事件流过/一个被处理多次等..
我发布了一个新的后续问题，因为观察到的行为确实有点奇怪/与我的预期不同：stackoverflow.com/questions/46361358/…

标签： c# system.reactive reactive-programming rx.net

【解决方案1】：

正如@nikoniko 已经提到的，油门可以解决问题。

using System;
using System.Reactive.Linq;

namespace Printing {
class Program {
    static void Main(string[] args) {
        var source = Observable.Interval(TimeSpan.FromMilliseconds(333))
            .Do(i => Console.WriteLine($"new item: {i}"));
        var sampling = source.Throttle(TimeSpan.FromSeconds(1))
            .Do(i => Console.WriteLine($"sampled: {i}"));

        var subscription = sampling.Subscribe();

        Console.ReadLine();

        subscription.Dispose();

        Console.ReadLine();
    }
}

} 什么都没有，因为来自源的事件以两个高频率到达。但是如果源需要更多时间来传递元素，那么节流中给出的时间跨度：

using System;
using System.Reactive.Linq;

namespace Printing {
    class Program {
        static void Main(string[] args) {
            var source = Observable.Interval(TimeSpan.FromSeconds(1.2))
                .Do(i => Console.WriteLine($"{DateTime.Now.ToShortTimeString()}: new item: {i}"));
            var sampling = source.Throttle(TimeSpan.FromSeconds(1))
                .Do(i => Console.WriteLine($"{DateTime.Now.ToShortTimeString()}:  {i}"));

            var subscription = sampling.Subscribe();

            Console.ReadLine();

            subscription.Dispose();

            Console.ReadLine();
        }
    }
}

限制时间结束后会出现结果。如您所见，在触发源中的事件后的第二秒，它将出现在结果中。

08:32:26: new item: 0
08:32:27: throttle 0
08:32:28: new item: 1
08:32:29: throttle 1
08:32:30: new item: 2
08:32:31: throttle 2
08:32:32: new item: 3
08:32:33: throttle 3
08:32:34: new item: 4
08:32:35: throttle 4
08:32:36: new item: 5
08:32:37: throttle 5

【讨论】：