仍然不明白 RXJS 运算符 mergeLatestAll 是如何工作的？

Question

我编写了以下示例，但仍然无法理解输出！

import { combineLatestAll, of, map } from 'rxjs';

const s1 = of(1, 2, 3).pipe(
  map(v1 =>
    of(100, 101, 102, 103, 104).pipe(
      map(v2 => [v1,v2])
    )
  ),
  combineLatestAll()
);

s1.subscribe(console.log);

输出是：

[ [ 1, 104 ], [ 2, 104 ], [ 3, 100 ] ]
[ [ 1, 104 ], [ 2, 104 ], [ 3, 101 ] ]
[ [ 1, 104 ], [ 2, 104 ], [ 3, 102 ] ]
[ [ 1, 104 ], [ 2, 104 ], [ 3, 103 ] ]
[ [ 1, 104 ], [ 2, 104 ], [ 3, 104 ] ]

我已经查看输出很长一段时间了，但仍然无法理解这个运算符的行为。

Answer 1

在理解之前combineLatestAll，我们先分析一下 的行为combineLatest。

根据此处显示的弹珠图，如果我们有 2 个数据流，则combineLatest“组合”这 2 个数据流中的最新值。

这意味着，在它订阅的 2 个流中，除非它们都开始发出数据，否则您将看不到任何输出。

但是，如果第一个流在第二个流开始时已经完成，会发生什么。

var c1 = of(5, 15, 25);
var c2 = of(6, 16, 26);

var s2 = combineLatest([c1,c2]);
s2.subscribe(console.log) // Outputs = [25,6] [25,16] and [25, 26]

在这种情况下，combineLatest运算符将从源 1 获取最新/最后一个值（上例中为 25），然后在第二个流启动时开始发出输出（组合来自源 1 的最新/最后一个值和来自源 2 的所有传入值）

如果有 3 个来源怎么办？在这种情况下，combineLatest仅当所有 3 个流都开始生成输出时才会生成输出。

var c1 = of(5, 15, 25);
var c2 = of(6, 16, 26);
var c3 = of(7, 17, 27);

var s2 = combineLatest([c1,c2, c3]);
s2.subscribe(console.log) // Outputs = [25,26, 7] [25,26, 17] and [25, 26, 27]

在上述情况下，当第三个 observable 开始发出值时，其他 2 个 observable 已经完成。因此，输出类似于从前两个源获取最新/最后一个值，然后将其与第三个可观察值的所有值相结合。

上面的例子也可以写成——

var s3 = of(c1, c2, c3).pipe(combineLatestAll());
s3.subscribe(console.log);

在您最初的问题中，您看到了完全相同的行为。前 2 个可观察值具有最新值，然后它与最后一个可观察值的每个值相结合。