RxJS （即响应式编程的实现库）是一个流行的 JavaScript 库，可以帮助程序员更轻松地处理异步数据流的代码。它提供了一系列的 API，允许我们处理多个可观察对象。在本文中，我们将介绍一些 RxJS 中用于合并和恢复多个流的方法，并提供一些实用的示例代码。
合并许多流
combineLatest 方法可用于合并多个流。尽管它不是仅限于要合并两个流，但我们将仅使用两个流来介绍此方法的工作原理。此方法将接收两个可观察的流，并在每个流发出最新值的情况下将这些流合并为一个流。以下为示例代码：
const streamOne$ = Rx.Observable.interval(1000); const streamTwo$ = Rx.Observable.interval(2000); Rx.Observable.combineLatest( streamOne$, streamTwo$ ).subscribe(val => console.log(val));
代码中的 interval 方法用于创建定期发出数字值的可观察流。在本例中，streamOne$ 每一秒发出一个值，而 streamTwo$ 每两秒发出一个值。combineLatest 方法将这两个流合并为一个，它也每当其中一个任何一个流发出值时就把值推送到流中。在这种情况下，模拟值将分别发送给两个流，结果将是 [0, 0]，[0, 1]，[1, 1]，[2, 1] 等。该方法可以以相同方式用来合并更多的流。
Zip 方法
zip 方法可以和 combineLatest 方法一样的工作，然而不同的是，它只会响应该合并的流中发射值对应的位置相同的情况。以下为示例代码：
const streamOne$ = Rx.Observable.interval(1000); const streamTwo$ = Rx.Observable.interval(2000); Rx.Observable.zip( streamOne$, streamTwo$ ).subscribe(val => console.log(val));
本例中， combineLatest 方法的输出值是 [0, 0]， [0, 1]， [1, 1]， [2, 1] 等，而在输出值上 zip 的输出值是 [0, 0]，[1, 1] 等。在本例中，zip 只有在它收到的是流发射了相应的位置相同的值后才会发出值。
ForkJoin 方法
forkJoin 方法是将多个流组合为一个口令的一种方式。在所有流发出一个值后，它会发出一个组合值。本方法只适用于除了异步之外还可以在不同的时间点引用所有的流，并且该流能够在请求结束后就停止发出值的情况下。以下为示例代码：
-- -------------------- ---- ------- ----- ---------- - -------------------------- ----- ---------- - -------------------------- ----- ------------ - -------------------------- ----------------------- ----------- ----------- ------------ --------------- -- ------------------
在本例中，我们使用了 timer 方法来创建一个发射单个值的可观察流。每个流都在不同的时间内发出一个值。在这种情况下，forkJoin 操作将在所有流发出值之后的第4秒后发出组合值。
Concat 方法
concat 方法可以用于按顺序的连接两个流。此方法将等待第一个流终止并触发“完结”事件，然后才会观察和发生下一个流里的事件流。以下为示例代码：
const streamOne$ = Rx.Observable.of('Hello');
const streamTwo$ = Rx.Observable.of('World!');
Rx.Observable.concat(
streamOne$,
streamTwo$
).subscribe(val => console.log(val));在本例中，我们使用 of 方法创建两个发出固定值的可观察流。 concat 方法将等待第一个流发出 & ldquo; Hello & rdquo; 的值，然后才会看到发出“ World! ”的值的第二个流中的事件流。
Merge 方法
merge 方法可以用于合并流，无论流何时发出一个值。这是最简单的合并流的方式之一。以下为示例代码：
const streamOne$ = Rx.Observable.interval(1000); const streamTwo$ = Rx.Observable.interval(2000); Rx.Observable.merge( streamOne$, streamTwo$ ).subscribe(val => console.log(val));
在本例中，我们使用 interval 方法创建两个可观察流。此方法将合并这两个流，并且无论流何时发出一个值，将值推送到合并后的流中。
取消合并流
switchMap 方法可以用于取消合并的流。它订阅新返回的函数并取消先前流的订阅。以下为示例代码：
const streamOne$ = Rx.Observable.interval(1000); const toggle$ = Rx.Observable.fromEvent(document, 'click'); streamOne$ .switchMap(() => toggle$) .subscribe(val => console.log(val));
在本例中， interval 方法将创建一个每秒发出一个值的可观察流，fromEvent 方法在事件（单击）触发时将创建一个新流。switchMap 方法在每次 interval 流发出值时将首先取消前一个 toggle$ 流的订阅，并使用一个新的 toggle$ 流。如果不使用 switchMap 方法，则会创建许多事件流并消耗许多资源。
结论
此处提供的每个方法都可以帮助您实现合并多个流或从多个流中恢复值。通过 API，可以动态地创建可观察的流并绘制数据流。因此， RxJS 被认为是 JavaScript 语言开发人员必须掌握的知识之一，无论是前端开发还是后端开发。
来源：JavaScript中文网 ，转载请注明来源 https://www.javascriptcn.com/post/672433972e7021665e1299e2