反应式扩展 SelectMany 和 Concat

我理解 SelectMany 的行为是有效地将生成的每个值的结果合并到单个流中，因此排序是不确定的。

如何在 C# 中的 RxJs 中执行类似于 concatAll 的操作。

var obs = Observable.Range (1, 10).SelectMany (x => {
return Observable.Interval (TimeSpan.FromSeconds(10 - x)).Take (3);
}).Concat();

这实际上是我想要做的，给定一个范围，等待一段时间，然后按照它们开始的顺序连接。显然这是一个玩具示例，但想法就在那里。

Blair

Use Select, not SelectMany. The Concat您想要使用的重载适用于IObservable<IObservable<T>>，所以只需投影内部序列，不要展平它们。

var obs = Observable.Range(1, 10)
                    .Select(x => Observable.Interval(TimeSpan.FromSeconds(10 - x)).Take(3))
                    .Concat();

请注意，每个人的订阅Interval通过使用推迟Concat;即第一个Interval当您订阅时立即开始，但所有剩余的间隔都会在没有订阅的情况下生成并排队。这不像Concat将订阅所有内容，然后以正确的顺序重播这些值。