反应式扩展：节流/样本具有不同的间隔答案

【问题标题】：Reactive Extensions: Throttle/Sample with varying interval反应式扩展：节流/样本具有不同的间隔
【发布时间】：2010-08-16 03:42:21
【问题描述】：

我有一个IObservable，它以随机间隔产生值，我想限制这个序列。我发现的一件事是Throttle 运营商对“节流”的定义与我的不同。

Throttle 仅在指定的时间间隔过后无声产生值（它产生最后看到的值）。我认为节流意味着在指定的时间间隔内产生值（当然，除非有静默）。

说，我希望 Observable.Interval(100).Select((_,i) => i).Throttle(200) 产生（以任何性能/时序问题为模）偶数，因为我将其限制为“半速”。然而，这个序列根本没有产生任何价值，因为从来没有一段长度为 200 的静默期。

所以，我发现Sample 实际上做了我想要的“节流”行为。 Observable.Interval(100).Select((_,i) => i).Sample(200) 产生（同样，以任何性能/时序问题为模）偶数序列。

但是，我还有另一个问题：间隔会有所不同，具体取决于最后一个“采样”值。我想要的是编写一个看起来像这样的运算符：

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector);

intervalSelector 参数产生下一个样本的间隔，而第一个样本……要么取自第一个值，要么取自附加参数，我不在乎。

我试着写这个，但我最终得到了一个大而复杂的结构，它不能很好地工作。我的问题是，我可以使用现有的运营商（也就是单线）来构建它吗？

【问题讨论】：

标签： .net system.reactive

【解决方案1】：

几个小时后，我睡了一会儿，我明白了。

public static IObservable<T> Sample<T>(this IObservable<T> source, Func<T, TimeSpan> intervalSelector)
{
    return source.TimeInterval()
                 .Scan(Tuple.Create(TimeSpan.Zero, false, default(T)), (acc, v) =>
                 {
                     if(v.Interval >= acc.Item1)
                     {
                         return Tuple.Create(intervalSelector(v.Value), true, v.Value);
                     }
                     return Tuple.Create(acc.Item1 - v.Interval, false, v.Value);
                 })
                 .Where(t => t.Item2)
                 .Select(x => x.Item3);
}

这按我的意愿工作：每次它产生一个值x，它会停止产生值，直到intervalSelector(x) 时间过去。

【讨论】：

【解决方案2】：

你看的不是for Observable.BufferWithTime吗？

【讨论】：

BufferWithTime 和其他的有同样的缺陷：时间间隔是固定的。我需要计算从最后一个采样值中获取下一个样本要等待多长时间。我看看能不能画个弹珠图……

【解决方案3】：

我以另一种方式做到了，我想与任何有用的人分享。

var random = new Random();
Observable.Return(Unit.Default)
            .SelectMany(_ => Observable.Timer(TimeSpan.FromSeconds(random.Next(1, 6))))
            .TimeInterval()
            .Do(value => Console.WriteLine(value))
            .Repeat()
            .Subscribe();

【讨论】：