CombineLatest 是否保留了 observables 的顺序？

Question

我对以下重载感兴趣：

public static IObservable<IList<TSource>> CombineLatest<TSource>(this params IObservable<TSource>[] sources);
public static IObservable<IList<TSource>> CombineLatest<TSource>(this IEnumerable<IObservable<TSource>> sources);

结果列表中元素的顺序是否保证与输入中的顺序相同？

例如，在下面的代码中，list[0] 将始终包含来自a 的元素和list[1] 来自b 的元素？

IObservable<int> a = ...;
IObservable<int> b = ...;
var list = await Observable.CombineLatest(a, b).FirstAsync();

文档说明：

包含最新源元素的列表

和：

包含源最新元素列表的可观察序列

但并没有真正提及任何关于订单的内容。

Answer 1

顺序是守恒的。

当您查看 source code of RX 时，一切都归结为 System.Reactive.Linq.CombineLatest<TSource, TResult> 类。

您可以在那里发现为每个输入 observable 创建了一个索引观察者（其中索引是输入中的顺序）：

for (int i = 0; i < N; i++)
{
    var j = i;

    var d = new SingleAssignmentDisposable();
    _subscriptions[j] = d;

    var o = new O(this, j);
    d.Disposable = srcs[j].SubscribeSafe(o);
}

生成的元素如下：

private void OnNext(int index, TSource value)
{
    lock (_gate)
    {
        _values[index] = value;
        _hasValue[index] = true;

        if (_hasValueAll || (_hasValueAll = _hasValue.All(Stubs<bool>.I)))
        {
                /* snip */
                res = _parent._resultSelector(new ReadOnlyCollection<TSource>(_values));
                /* snip */

            _observer.OnNext(res);
        }
        /* snip */
    }
}

我感兴趣的重载的_resultSelector 只是Enumerable.ToList()。所以输出列表中的顺序将与输入中的顺序相同。

Answer 2

CombineLatest 在有元素推送到所有流时首先触发。

之后，每当有新元素被推送到任何流时，它都会触发。

因此，如果您“组合”两个流，则结果列表中始终包含两个元素，据我所知，可以保证顺序与您在 CombineLatest 参数中给出的流的顺序相同。

您可以使用大理石图来可视化 Rx 运算符。 HERE 是 CombineLatest 的那个。