如何创建一个 IObservable<T> 每 -n- 秒返回一个值而不跳过任何答案

【问题标题】：How do I create an IObservable<T> that returns a value every -n- seconds without skipping any如何创建一个 IObservable<T> 每 -n- 秒返回一个值而不跳过任何
【发布时间】：2010-10-04 20:38:22
【问题描述】：

下面这个例子是我尝试这样做的：

var source
    = Observable.Sample(
          Observable.Range(1, int.MaxValue), TimeSpan.FromSeconds(2));

但是当我 .Subscribe() 到那个 Observable 并将它输出到控制台时，它会显示一个这样的序列，每 2 秒输出一行：

OnNext: 312969
OnNext: 584486
OnNext: 862009

显然 .Range() 可观察对象正在运行，而 .Sample() 可观察对象在每次输出之间等待 2 秒。我想知道如何创建一个 observable 但不允许跳过值，所以很明显它看起来像这样：

OnNext: 1
OnNext: 2
OnNext: 3

每 2 秒输出一个来自 .Range() 的值。如何在 .NET 的响应式扩展中实现这一点？

【问题讨论】：

标签： c# system.reactive

【解决方案1】：

使用 Observable.GenerateWithTime：

var source = Observable.GenerateWithTime(1, _ => true, x => ++x, x => x, x => TimeSpan.FromSeconds(2));

Observable.Range 使用 Observable.Generate，所以这是一种方法。可能还有很多其他方法。

对于更高级的内容，例如以相同方式处理事件（因为这显然只有在您自己生成数据时才有帮助），请参阅处理此问题并已解决的 How to throttle event stream using RX?。

【讨论】：

谢谢，这也是一个有趣的了解方法。

【解决方案2】：

我最近通过创建Observable that emits timed events every timeInterval 来解决这个问题。然后，您可以使用 Zip 方法将 Observable 中的事件与计时器 Observable 中的事件同步。

例如：

    var timer = 
        Observable
            .Timer(
                TimeSpan.FromSeconds(0), 
                TimeSpan.FromSeconds(2)
            );
    var source = Observable.Range(1, int.MaxValue);
    var timedSource = source.Zip(timer,(s,t)=>s);
    timedSource.Subscribe(Console.WriteLine);

【讨论】：

效果很好，谢谢。我把它包装成我自己的扩展方法 Pace(..)
在这种情况下，源值将被缓存并且可能会抛出 OOM 异常（afaik）

【解决方案3】：

一个众所周知的起搏器示例：

public static IObservable<T> Pace<T>(this IObservable<T> source, Timespan interval) =>
source
  .Select(p =>
    Observable
     .Empty<T>()
     .Delay(interval)
     .StartWith(p)
  )
  .Concat();

【讨论】：

您可以使用Return 方法来代替Empty+StartWith。 Pace 操作符的问题在于，如果source 序列比interval 更频繁地发出值，您最终会出现严重的内存泄漏。
是的，这对任何开发者来说都应该是显而易见的——如果你有一个独立的发射器——任何爆裂平坦化器都可能导致内存问题