.net - 如何使用Rx(可扩展功能)在Observable序列中粘贴相对延迟

Question

我开始使用Reactive扩展(版本2.1，以防万一)进行开发，对于我的示例应用程序，我需要按一定间隔(即每1秒)推送一系列int值。

我知道，我可以使用Observable.Range<int>(0,10)创建一个序列，但是我不知道如何设置两次推送之间的相对时间。我尝试过Delay()，但一开始只将序列移位一次。

然后，我找到了Observable.Generate()方法，可以通过以下方式将其调整为该任务:

var delayed = Observable.
              Generate(0, i => i <= 10, i => i + 1, i => i,
                          i => TimeSpan.FromSeconds(1));

但这似乎仅适用于简单的“for-each-like”定义的序列。
因此，总的来说，我的问题是，我们是否可以获取任何源序列并将其包装到某些代理中，这些代理将从源中提取消息并随着时间延迟将其进一步推送？

S--d1--d2--d3--d4--d5-|
D--d1-delay-d2-delay-d3-delay-d4-delay-d5-|

P.S. 如果这种方法与ReactiveExtensions的概念相矛盾，请也注意这一点。我不想“一定”这样做，将来他们还会遇到其他一些设计问题。

P.P.S 总体思路是，即使输入序列是有限的还是无限的，以及它推送事件的频率，也要确保 输出序列在事件之间具有指定的间隔。

Answer 1

Observable.Interval是您要查看的内容。它将生成一个基于0的长值，该长度在您指定的每个时间间隔内增加1，例如:

Observable.Interval(TimeSpan.FromSeconds(1)).Subscribe(x => Console.WriteLine(x));

然后，您可以根据需要使用投影( Select)偏移/更改此值。

您还可以使用Zip运算符将一个流与另一个流“并排”-您可能也想看看。 Zip将来自两个流的事件配对在一起，因此它以当前最慢的流的速度发出。 Zip也非常灵活，它可以压缩任意数量的流，甚至可以将IObservable压缩为IEnumerable。这是一个例子:

var pets = new List<string> { "Dog", "Cat", "Elephant" };
var pace = Observable.Interval(TimeSpan.FromSeconds(1))
    .Zip(pets, (n, p) => p)     
    .Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("Done"));

这会每隔1秒钟将宠物写出一次。

鉴于P.P.S.在上面添加的内容中，我将给出另一个答案-因为无论如何它都是一种有用的技术，我将留作引用。