如何在长时间运行的查询中延长限制时间跨度

Question

是否可以在查询过程中扩展限制时间跨度值？例如，假设在 101 Rx 样本限制中有一个示例，则有此查询var throttled = observable.Throttle(TimeSpan.FromMilliseconds(750));。

如果我想更改它，以便在前 500 毫秒内没有事件，那么限制值将扩展到，例如，此后每个事件的 1500 毫秒。

这会是使用 Switch 运算符的地方吗？

Answer 1

有一个重载 Throttle 接受一个工厂函数，该函数接受源事件并生成一个IObservable<T>的"限制流"（T 可以是任何类型）。事件将被抑制，直到油门流发出。

以下示例有一个每秒泵送一次的流，节流工厂产生 0.5 秒的节流。因此，在开始时，源流不会受到限制。

如果您输入 2，则油门将更改为两秒油门，并且所有事件都将被抑制。向下更改为 1，事件将再次出现。

void Main()
{
    var throttleDuration = TimeSpan.FromSeconds(0.5);
    Func<long, IObservable<long>> throttleFactory =
        _ => Observable.Timer(throttleDuration);
    var sequence = Observable.Interval(TimeSpan.FromSeconds(1))
                             .Throttle(throttleFactory);
    var subscription = sequence.Subscribe(Console.WriteLine);
    string input = null;
    Console.WriteLine("Enter throttle duration in seconds or q to quit");
    while(input != "q")
    {       
        input = Console.ReadLine().Trim().ToLowerInvariant();
        double duration;
        if(input == "q") break;
        if(!double.TryParse(input, out duration))
        {
            Console.WriteLine("Eh?");
            continue;
        }
        throttleDuration = TimeSpan.FromSeconds(duration);
    }
    subscription.Dispose();
    Console.WriteLine("Done");
}

由于这是一个为每个事件生成限制的工厂函数，因此您可以创建更动态的内容，以根据特定输入事件返回限制流。

像这样将流用作控件的想法是整个 Rx API 中使用的一种非常常见的技术，非常值得您思考：类似用途的示例包括 TakeUntil 的 other 参数、GroupByUntil 中的durationSelector、Buffer 中的bufferClosingSelector。