通过超级直观的图表学习合并Rxjs

内容来自于Max Koretskyi aka Wizard的《Learn to combine RxJs sequences with super intuitive interactive diagrams》

在足够复杂的应用程序上工作时，通常会有来自多个数据源的数据。它可以是一些多个外部数据点。序列合成是一种技术，通过将相关流组合成一个流，可以跨多个数据源创建复杂的查询。RxJs提供了各种各样的操作符，可以帮助你做到这一点，在本文中，我们将看看最常用的操作符。

下面是我将在接下里的文章里将会用到的图表的样例：

同时合并多个序列

我们将要看到的第一个操作符是merge。该运算符将多个可观察流组合在一起，同时从每个给定的输入流中发出所有值。当所有组成这个流的输入流产生值的时候，这些值都会作为合成流的结果被发出。这个过程在文档中经常被称为扁平化。

当所有输入流都结束了，那这个流就结束了。任何一个输入流引发了错误，则这个流引发错误。只要有一个流没有完成，则这个流就不会完成。
如果您不关心排放顺序，只关心来自多个组合流的所有值，就像它们是由一个流产生的一样，请使用此运算符。

在下图中，你可以看到merge合并了A,B两个流，每一个流都产生了3个值，当值被发出的时候，值会落入合成流中，最终由合成流发出。

下面是演示代码：

const a = stream(‘a’, 200, 3, ‘partial’);
const b = stream(‘b’, 200, 3, ‘partial’);
merge(a, b).subscribe(fullObserver(‘merge’));

// can also be used as an instance operator
a.pipe(merge(b)).subscribe(fullObserver(‘merge’));

顺序连接多个序列

接下来我们要讲到的操作符是concat。它将所有的输入流串联起来，顺序的订阅并发送每一个流的值。一旦当前流完成，它会订阅下一个流，并将输入流发出的值传递到结果流中。

当所有输入流完成时，该流完成，如果某些输入流引发错误，将引发错误。如果一些输入流没有完成，它将永远不会完成，这也意味着一些流将永远不会被订阅。

如果排放顺序很重要，并且您希望首先看到由您首先传递给操作符的流发送的值，请使用此运算符。例如，您可能有一个从缓存传递值的可观察序列和另一个从远程服务器传递值的序列。如果您想要合并它们并确保首先传递来自缓存的值，请使用concat。

在下图中，您可以看到concat运算符将两个流A和B组合在一起，每个流产生3个值，值首先从A开始，然后从B开始，一直到结果流。

下面是演示代码：

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 200, 3, 'partial');
concat(a, b).subscribe(fullObserver('concat'));
// can also be used as an instance operator
a.pipe(concat(b)).subscribe(fullObserver(‘concat’));

多个流竞争

接下来我们要讲到的这个操作符race，相当的有趣。它并不是将多个输入流合成一个流输出，而是多个流竞争，一旦有一个输入流最先发出值，那其他流将被取消订阅并完全忽略。

当选定的输入流完成时，结果流完成，如果这个流出错，将抛出一个错误。如果内部流不完成，它也永远不会完成。

如果你有多个可以提供价值的资源，例如世界各地的服务器，该运算符可能会很有用，但是由于网络条件的原因，延迟是不可预测的，并且变化很大。使用这个运算符，你可以将同一个请求发送到多个数据源，并使用第一个响应的结果。

在下图中，您可以看到race操作符将两个流A和B组合在一起，每个流产生3个项目，但是只有流A中的值被发出，因为这个流首先开始发出值。

下面是演示代码：

const a = intervalProducer(‘a’, 200, 3, ‘partial’);
const b = intervalProducer(‘b’, 500, 3, ‘partial’);
race(a, b).subscribe(fullObserver(‘race’));
// can also be used as an instance operator
a.pipe(race(b)).subscribe(fullObserver(‘race’));

组合为止数量的流和高阶可观察对象

之前讲到的操作，都只能组合已知数量的流。但是如果您事先不知道所有的流，并且想要合并可以在运行时延迟评估的流，会怎么样呢？事实上，这是使用异步代码时非常常见的情况。例如，对某些资源的网络调用可能会导致由原始请求的结果值决定的许多其他请求。

RxJs有我们在上面看到的操作符的变体，这些操作符采用一系列序列，被称为高阶Observable或Observable。

MergeAll

该运算符组合所有发出的内部流，就像普通合并一样，同时从每个流中生成值。

在下图中，你将看到一个高阶流H，它发出两个内部类A和B。mergeAll运算符将这两个流中的值组合起来，然后在它们发出值时将它们传递给结果流。

下面是演示代码：

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));

h.pipe(mergeAll()).subscribe(fullObserver(‘mergeAll’));

ConcatAll

该运算符将所有发出的内部流组合起来，就像普通concat一样，从每个流中顺序生成值。在下图中，您可以看到产生两个内部流A和B的高阶流H。串联运算符首先从A流中获取值，然后从流B中获取值，并将它们传递给结果序列。

下面是演示代码：

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));
h.pipe(concatAll()).subscribe(fullObserver(‘concatAll’));

SwitchAll

有时从所有内部Observable中接收值不是我们需要的。在某些情况下，我们可能只对最新内部序列中的值感兴趣。搜索功能是一个很好的例子。
当用户在输入框输入一些值后，我们想服务器发送一些网络请求，但这些网络请求是异步的。如果用户在返回结果之前又更新了输入框中的值，会发生什么？第二个网络请求被发送了出去，所以现在我们已经向服务器发送了两个搜索的网络请求。然而，我们对第一次搜索的结果已经不感兴趣了，并且，如果将两次搜索结果都显示给用户，这将不符合我们的设想。所以我们使用switchAll操作符，它只会订阅最新的内部流并产生值，并忽略之前的流。

在下图中，您可以看到产生两个内部流A和B的高阶流H。开关操作符首先从A流中获取值，然后从B流中获取值，并将它们传递给结果序列。

下面是演示代码：

const a = stream(‘a’, 200, 3);
const b = stream(‘b’, 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));

h.pipe(switchAll()).subscribe(fullObserver(‘switchAll’));

concatMap,mergeMap,switchMap

有趣的是，这些映射操作符concatMap,mergeMap,switchMap的使用频率比和他们相对应的concatAll,'mergeMap',switchAll要高得多。然而，如果你仔细想想，它们几乎是一样的。所有的*Map操作符都是由两个parts — producing流通过映射和使用组合逻辑，在由高阶Observable产生的内部流上进行观察。

让我们来看看下面熟悉的代码，它演示了mergeAll运算符是如何工作的:

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), map(i => [a, b][i]));

h.pipe(mergeAll()).subscribe(fullObserver('mergeAll'));

这里的map操作符产生Observable流，mergeAll合并这些Observable流。所以我们可以使用mergeMap轻松替代mergeAll。

const a = stream('a', 200, 3);
const b = stream('b', 200, 3);
const h = interval(100).pipe(take(2), mergeMap(i => [a, b][i]));

h.subscribe(fullObserver('mergeMap'));

这两个结果是完全一样的。concaMap和switchMap操作也是如此。你可以自己尝试一下。

配对序列组合

前面的操作符允许我们展平多个序列，并通过结果流不变地传递来自这些序列的值，就好像它们都来自这个序列一样。接下来我们要看的这组运算符仍然将多个序列作为输入，但不同之处在于它们将每个序列的值配对，为输出序列产生一个组合值。

每个运算符都可以选择一个所谓的投影函数作为最后一个参数，该参数定义了结果序列中的值应该如何组合。在我的示例中，我将使用默认的投影函数，该函数简单地使用逗号作为分隔符来连接值。在这一节的最后，我将展示如何提供一个定制的投影函数。

CombineLatest

我们要看到的第一个操作符是combineLatest。它允许您从输入序列中获取最新的值，并将这些值转换为结果序列的一个值。RxJs缓存每个输入序列的最后一个值，一旦所有序列产生了至少一个值，它就使用从缓存中获取最新值的投影函数来计算结果值，然后通过结果流发出该计算的输出。如果任何一个内部流不完成，它将永远不会完成。另一方面，如果任何一个流不发出值而是完成了，则结果流将在同一时刻完成而不发出任何信号，因为现在不可能在结果序列中包含来自完成的输入流的值。此外，如果某个输入流不发出任何值并且永远不会完成，combineLatest也永远不会发出并且永远不会完成，因为它将再次等待所有流发出某个值。

如果您需要评估一些状态组合，而这些状态组合需要在部分状态发生变化时保持最新，那么这个运算符会很有用。一个简单的例子就是监控系统。每个服务都由一个返回布尔值的序列表示，该值指示所述服务的可用性。如果所有服务都可用，则监控状态为绿色，因此投影功能只需执行逻辑“与”。

在下图中，你可以看到combineLatest操作组合了两个流A和B。一旦所有流都发射了至少一个值，每个新发射通过结果流产生一个组合值。

下面是实例代码：

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

combineLatest(a, b).subscribe(fullObserver('latest'));

Zip

这个操作符也是一个非常有趣的合并操作符，它在某种程度上类似于衣服或袋子上拉链的机械结构。它将两个或多个相应值的序列集合成一个元组(在两个输入流的情况下是一对)。它等待从所有输入流中发出相应的值，然后使用投影函数将它们转换成单个值并发出结果。只有当每个源序列中有一对新值时，它才会发布，因此如果其中一个源序列发布值的速度快于另一个序列，发布速率将由两个序列中较慢的一个决定。

当任何内部流完成并且相应的匹配对从其他流发出时，结果流完成。如果任何内部流没有完成，它将永远不会完成，如果任何内部流出错，它将抛出一个错误。

该运算符可方便地用于实现一个流，该流产生一系列具有间隔的值。以下是投影函数仅从range流返回值的基本示例:

zip(range(3, 5), interval(500), v => v).subscribe();

在下图中，您可以看到zip运算符将两个流A和B组合在一起。一旦对应的流对匹配，结果序列就会产生一个组合值:

以下是示例代码：

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

zip(a, b).subscribe(fullObserver('zip'));

forkjoin

有时，您有一组流，只关心每个流的最终发射值。通常这种序列只有一次发射。例如，您可能希望发出多个网络请求，并且只希望在收到所有请求的响应后采取措施。在某种程度上，它类似于Promise.all的功能。但是，如果您有一个发出多个值的流，除了最后一个值之外，这些值将被忽略。

当所有内部流完成时，生成的流只发出一次。如果任何内部流没有完成，它将永远不会完成，如果任何内部流出错，它将抛出一个错误。

在下图中，您可以看到forkJoin运算符将两个流A和b组合在一起。一旦对应的流对匹配，结果序列就会产生一个组合值:

下面是示例代码：

const a = stream('a', 200, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

forkJoin(a, b).subscribe(fullObserver('forkJoin'));

WithLatestFrom

我们在本文中最后要看的运算符是withLatestFrom。当您有一个引导流，但还需要来自其他流的最新值时，使用该运算符。在某种程度上，它类似于combineLatest操作符，每当任何输入流有新的排放时，都会发出新的值。withLatestFrom只有在引导流发出值后，才会发出新值。

正如combineLatest一样，它仍然等待来自每个流的至少一个发射值，并且当引导流完成时，可以在没有单个发射的情况下完成。如果引导流没有完成，它将永远不会完成，如果任何内部流出错，它将抛出一个错误。

在下图中，您可以看到withLatestFrom运算符将两个流A和流B组合在一起，其中流B是引导流。每当流B发出一个新值时，产生的序列使用来自流A的最新值产生一个组合值。

下面是示例代码：

const a = stream('a', 3000, 3, 'partial');
const b = stream('b', 500, 3, 'partial');

b.pipe(withLatestFrom(a)).subscribe(fullObserver('latest'));

Projection function(投影函数)

如本节开头所述，通过配对组合值的所有运算符都采用可选的投影函数。该函数定义结果值的转换。使用此函数，您可以选择只从特定的输入序列中发出一个值，或者以任何您想要的方式连接值:

// return value from the second sequence
zip(s1, s2, s3, (v1, v2, v3) => v2)

// join values using dash as a separator
zip(s1, s2, s3, (v1, v2, v3) => `${v1}-${v2}-${v3}`)

// return single boolean result
zip(s1, s2, s3, (v1, v2, v3) => v1 && v2 && v3)