RxJS 中的 merge 操作符详解

在 RxJS 中，merge 操作符是一种非常重要且常用的操作符。它可以将多个 Observable 序列合并为一个单独的 Observable 序列，并发射它们所发射的所有项。

本文将详细介绍 merge 操作符的使用方法，让你了解它的内部实现原理，并提供一些示例代码以帮助你更好地理解。

RxJS 的 merge 操作符如何使用

使用 RxJS 中的 merge 操作符非常简单。代码如下：

import { merge } from 'rxjs';

const source1$ = of('Hello');
const source2$ = of('World');

const merged$ = merge(source1$, source2$);

merged$.subscribe(console.log);

在上面的代码中，我们使用 import 语句从 RxJS 库中导入 merge 操作符。接着，我们创建了两个 Observable 序列：source1$ 和 source2$。然后，我们将这两个 Observable 序列传递给 merge 操作符，使用它来创建一个新的 Observable 序列 merged$。

最后，我们使用 merged$ 的 subscribe() 方法来订阅这个 Observable 序列，并将它的每一个发射值打印到控制台上。

merge 操作符发射值的顺序

当多个 Observable 序列被合并时，它们的发射值的顺序通常是由它们发射的时间决定的。具体地，merge 操作符将同时订阅每个 Observable 序列，并在它们发射值的同时将它们合并到一个新的 Observable 序列中。

下面的代码演示了 merge 操作符如何按照发射值的时间顺序合并两个 Observable 序列。

import { interval, merge } from 'rxjs';
import { mapTo } from 'rxjs/operators';

const interval1$ = interval(500).pipe(mapTo('Hello'));
const interval2$ = interval(1000).pipe(mapTo('World'));

const merged$ = merge(interval1$, interval2$);

merged$.subscribe(console.log);

在上面的代码中，我们创建了两个 Observable 序列 interval1$ 和 interval2$。interval1$ 每隔 500 毫秒发射一个 'Hello' 值，interval2$ 每隔 1000 毫秒发射一个 'World' 值。然后，我们使用 merge 操作符将它们合并在一起，得到一个新的 Observable 序列 merged$。

当我们对 merged$ 进行订阅时，它会依次发射 'Hello' 和 'World' 值。具体来说，第一个 'Hello' 值会在订阅后的 500 毫秒就被发射出来，随后是两个 'World' 值，分别在订阅后的 1000 毫秒和 1500 毫秒被发射出来。最后，merged$ 发射完成事件。

merge 操作符如何处理错误

在 RxJS 中，当一个 Observable 序列发生错误时，它将停止发射新的值，并通过它的错误处理函数来通知它的订阅者。当多个 Observable 序列被合并时，如果其中某一个序列发生错误，那么 merge 操作符将停止所有的 Observable 序列，并通过合并后的 Observable 序列的错误处理函数来通知它的订阅者。

下面的代码演示了 merge 操作符如何处理来自多个 Observable 序列的错误。

import { throwError, of, merge } from 'rxjs';

const source1$ = of('Hello');
const source2$ = throwError('Error Occurred');

const merged$ = merge(source1$, source2$);

merged$.subscribe({
  next: console.log,
  error: console.error,
  complete: () => console.log('Complete')
});

在上面的代码中，我们创建了两个 Observable 序列：source1$ 和 source2$。其中 source1$ 发射一个 'Hello' 值，source2$ 则发射一个错误。然后，我们使用 merge 操作符将它们合并在一起，得到一个新的 Observable 序列 merged$。

我们在 merged$ 的 subscribe() 方法中传递了一个对象，它的 next 回调函数用来处理 merged$ 发射的值，error 回调函数用来处理 merged$ 发射的错误，complete 回调函数用来处理 merged$ 发射完成事件。

当我们对 merged$ 进行订阅时，即使 source2$ 发射了一个错误，merged$ 仍然可以发射 source1$ 的 'Hello' 值，并将错误通过 error 回调函数传递给我们。最后，merged$ 发射完成事件，并通过 complete 回调函数通知我们。

merge 操作符的其他参数

RxJS 的 merge 操作符还有一些可选参数，可以让我们更加灵活地使用它。下面是一些常用的参数：

concurrent：控制同时订阅的 Observable 序列的数量

当你要合并一组较大的 Observable 序列时，可能会遇到性能问题。特别是当你的 Observable 序列是网络请求或者文件读取等 IO 操作时，它们的订阅和释放需要一定的时间和资源。RxJS 的 merge 操作符提供了 concurrent 参数，可以让你控制同时订阅的 Observable 序列的数量，从而避免资源的浪费。

下面的代码演示了如何使用 concurrent 参数。

import { interval, merge } from 'rxjs';
import { take, mapTo } from 'rxjs/operators';

const interval1$ = interval(500).pipe(take(3), mapTo('Source 1'));
const interval2$ = interval(1000).pipe(take(3), mapTo('Source 2'));
const interval3$ = interval(1500).pipe(take(3), mapTo('Source 3'));

const merged$ = merge(interval1$, interval2$, interval3$, 2);

merged$.subscribe(console.log);

在上面的代码中，我们创建了三个 Observable 序列：interval1$、interval2$ 和 interval3$。它们分别每隔 500、1000 和 1500 毫秒发射一个值，而每个序列只能发射三个值，然后就完成了。注意这里我们使用了 take 操作符来限制每个序列的发射数量。

我们将这三个 Observable 序列传递给 merge 操作符，并指定 concurrent 参数为 2。这意味着，只有同时订阅两个 Observable 序列，而第三个序列需要等到前面的序列发射完毕后才能开始订阅。

当我们对 merged$ 进行订阅时，会依次发射 'Source 1'、'Source 2'、'Source 1'、'Source 3'、'Source 2'、'Source 3' 几个值。具体来说，'Source 1' 值在订阅后的 500 毫秒被发射出来，'Source 2' 值在订阅后的 1000 毫秒被发射出来，'Source 3' 在订阅后的 1500 毫秒被发射出来。

Scheduler：控制合并的操作运行在哪个调度器上

RxJS 的 merge 操作符也支持使用 Scheduler 参数来控制合并的操作运行在哪个调度器上。如果你在创建 Observable 序列时使用了不同的调度器，那么可以使用 Scheduler 参数来统一它们的调度器。

下面的代码演示了如何使用 Scheduler 参数。

import { interval, merge, asapScheduler } from 'rxjs';
import { take, mapTo, observeOn } from 'rxjs/operators';

const interval1$ = interval(500, asapScheduler).pipe(take(3), mapTo('Source 1'));
const interval2$ = interval(1000).pipe(take(3), mapTo('Source 2'));
const interval3$ = interval(1500).pipe(take(3), mapTo('Source 3'));

const merged$ = merge(
  interval1$.pipe(observeOn(asapScheduler)),
  interval2$,
  interval3$,
  undefined,
  asapScheduler
);

merged$.subscribe(console.log);

在上面的代码中，我们创建了三个 Observable 序列：interval1$、interval2$ 和 interval3$。它们分别每隔 500、1000 和 1500 毫秒发射一个值，而每个序列只能发射三个值，然后就完成了。

注意这里，我们在创建 interval1$ 序列时，使用了 asapScheduler 调度器。asapScheduler 调度器会尽快将观察者的执行放到 JavaScript 事件循环的队列中。而 interval2$ 和 interval3$ 序列则使用默认的 asyncScheduler 调度器。

我们使用 observeOn 操作符将 interval1$ 序列的调度器切换到 asapScheduler。接着，我们使用 merge 操作符将这三个 Observable 序列合并在一起，并指定 Scheduler 参数为 asapScheduler。

当我们对 merged$ 进行订阅时，它会依次发射 'Source 1'、'Source 2'、'Source 3'、'Source 1'、'Source 2'、'Source 3' 几个值。注意这里，'Source 1' 值会在 'Source 2' 值之前发射，这是因为我们使用了 asapScheduler 调度器，这使得 interval1$ 序列的值更早地被发射了。

总结

在本文中，我们介绍了 RxJS 中的 merge 操作符。我们详细讲解了它的使用方法、发射值的顺序、错误处理和其他可选参数。我们提供了一些示例代码，帮助你更好地理解如何使用 merge 操作符来合并多个 Observable 序列。

在平时的开发工作中，掌握好 merge 操作符的使用方法，可以让我们更好地处理各种异步流程，解决效率和资源的问题，提高我们的代码质量和开发效率。

来源：JavaScript中文网 ，转载请注明来源 本文地址：https://www.javascriptcn.com/post/65b0772aadd4f0e0ff9d17aa