经过超级直观的图表学习合并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)