node.js - mergeMap 运算符究竟是如何工作的，在哪些情况下使用它？

Question

在来这里之前我已经阅读了 Rxjs 的官方文档和其他一些页面，但我仍然不清楚。我的理解是这样的：

它用于“加入” 2 个可观察对象，从而获得单个可观察对象，我还看到它用于“展平”一个可观察对象（我也不是很清楚）。

现在......我有几天尝试使用 Angular 和 Node.js 和 Express 来编写用户注册表，我找到了一个我决定使用的小教程，它有以下代码：

import { Injectable, Injector } from '@angular/core';
import { HttpClient, HttpInterceptor, HttpRequest, HttpHandler, HttpEvent, HttpErrorResponse } from '@angular/common/http';
import { Observable, throwError } from 'rxjs';
import { catchError, retry, mergeMap } from 'rxjs/operators'
import { AuthenticationService } from './authentication.service';

@Injectable({
	providedIn: 'root'
})
export class AppInterceptor implements HttpInterceptor {

	constructor(private injector: Injector) { }

	intercept(req: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
		let accessToken = "", refreshToken = ""

		const tokens = JSON.parse(sessionStorage.getItem("tokens"))
		if (tokens) {
			accessToken = tokens.accessToken
			refreshToken = tokens.refreshToken
		}

		let clonHttp: HttpRequest<any> 
		clonHttp = tokens ? req.clone({ headers: req.headers.append("Authorization", `Bearer ${accessToken}`) }) : req
		
		let auth = this.injector.get(AuthenticationService);

		return next.handle(clonHttp)
			.pipe(
				catchError((error: HttpErrorResponse) => {
					if (error.error instanceof ErrorEvent) {
						console.log("error event")
					} else if (error.status == 401) {
						return auth.getNewAccessToken(refreshToken)
							.pipe(
								retry(3),
								mergeMap(
									(response: any) => {
										tokens.accessToken = response.accessToken
										sessionStorage.setItem("tokens", JSON.stringify(tokens))

										clonHttp = req.clone({ headers: req.headers.append("Authorization", `Bearer ${response.accessToken}`) })
										return next.handle(clonHttp)
									}
								)
							)
					} else if (error.status == 409) {
						return throwError("User not logged")
					} else {
						if (error.error && error.error.message) {
							return throwError(error.error.message)
						} else {
							return throwError("Check your connection")
						}
					}
				})
			)
	}

}

如果您看到，当您使用 MergeMap 运算符时，它们只会将答案传递给您（一个可观察的），或者至少这是我能看到的。我想说的是，我没有看到他们将它与 2 个可观察对象一起使用或混合 2 个可观察对象，这是我在他们的官方文档中读到的，事实上，在他们展示的示例中，他们总是使用它有 2 个可观察的。

老实说，我很难理解这个运算符，如果有人可以帮助我以简单的方式理解它，除了理解它在我之前展示的代码中的用法外，我将非常感激。提前问好。谢谢！

标签： node.jsangularrxjsmergemap

Answer 1

mergeMap，像许多其他所谓的高阶映射运算符一样，维护一个或多个内部可观察对象。

使用外部值和提供的函数创建内部 observable。外部值本质上只是从其源接收到的值。例如：

of(1, 2, 3).pipe(
  mergeMap((outerValue, index) => /* ... return an observable ... */)
).subscribe(); // `outerValue`: 1, 2, 3 (separately)

当一个外部值进来时，一个新的内部 observable将被创建。我认为理解这一点的最好方法是查看源代码：

// `value` - the `outerValue`
protected _next(value: T): void {
  if (this.active < this.concurrent) {
    this._tryNext(value);
  } else {
    this.buffer.push(value);
  }
}

protected _tryNext(value: T) {
  let result: ObservableInput<R>;
  const index = this.index++;
  try {
    // Create the inner observable based on the `outerValue` and the provided function (`this.project`)
    // `mergeMap(project)`
    result = this.project(value, index);
  } catch (err) {
    this.destination.error(err);
    return;
  }
  this.active++;
  // Subscribe to the inner observable
  this._innerSub(result, value, index);
}

请暂时忽略concurrent，buffer我们稍后再看。

现在，当内部 observable 发出时会发生什么？在进一步讨论之前，值得一提的是，尽管很明显，内部可观察对象 需要内部订阅者。我们可以在_innerSub上面的方法中看到这一点：

private _innerSub(ish: ObservableInput<R>, value: T, index: number): void {
  const innerSubscriber = new InnerSubscriber(this, value, index);
  const destination = this.destination as Subscription;
  destination.add(innerSubscriber);
  // This is where the subscription takes place
  subscribeToResult<T, R>(this, ish, undefined, undefined, innerSubscriber);
}

当内部 observable 发出时，将调用该notifyNext方法：

notifyNext(outerValue: T, innerValue: R,
            outerIndex: number, innerIndex: number,
            innerSub: InnerSubscriber<T, R>): void {
  this.destination.next(innerValue);
}

目的地指向链中的下一个订阅者。例如，它可以是这样的：

of(1)
  .pipe(
    mergeMap(/* ... */)
  )
  .subscribe({} /* <- this is the `destination` for `mergeMap` */)

这将在下面的链中的下一个订阅者如何中更详细地解释。

那么，这意味着to mix 2 observables什么？

让我们看看这个例子：

of(2, 3, 1)
  .pipe(
    mergeMap(outerValue => timer(outerValue).pipe(mapTo(outerValue)))
  )
  .subscribe(console.log)
  /* 1 \n 2 \n 3 */

2到达时，将mergeMap订阅一个内部可观察对象，该可观察对象将在200毫秒内发出。这是一个异步操作，但请注意外部值 (2, 3, 1) 是同步到达的。接下来，3到达并将创建一个内部 obs。300这将在毫秒内发出。由于当前脚本尚未完成执行，因此尚未考虑回调队列。现在1到了，并将创建一个内部 obs。100这将在毫秒内发出。

mergeMap现在有 3 个内部可观察对象，并将传递任何内部可观察对象发出的内部值。
正如预期的那样，我们得到1, 2, 3.

所以就是mergeMap这样。混合 observable可以这样想：如果一个外部值出现并且一个内部 observable 已经创建，那么mergeMap简单地说：“没问题，我将创建一个新的内部 obs。并订阅它”。

关于concurrent和buffer

mergeMap可以给定第二个参数，concurrent它指示应该同时处理多少个内部可观察对象。active使用该属性跟踪这些活动的内部可观察对象的数量。

如方法中所见_next，如果active >= concurrent，outerValues将被添加到 a buffer，这是一个 queue( FIFO)。

然后，当一个活动的内部可观察对象完成时，mergeMap将从该值中获取最旧的值，并使用提供的函数从中创建一个内部可观察对象：

// Called when an inner observable completes
notifyComplete(innerSub: Subscription): void {
  const buffer = this.buffer;
  this.remove(innerSub);
  this.active--;
  if (buffer.length > 0) {
    this._next(buffer.shift()!); // Create a new inner obs. with the oldest buffered value
  } else if (this.active === 0 && this.hasCompleted) {
    this.destination.complete();
  }
}

考虑到这一点，concatMap(project)只是mergeMap(project, 1)。

所以，如果你有：

of(2, 3, 1)
  .pipe(
    mergeMap(outerValue => timer(outerValue * 100).pipe(mapTo(outerValue)), 1)
  )
  .subscribe(console.log)

这将被记录：

2 \n 3 \n 1.

链中的下一个订阅者呢

运算符是返回另一个函数的函数，该函数接受一个observable作为其唯一参数并返回另一个observable。当订阅流时，操作员返回的每个可观察对象都有自己的订阅者。

所有这些订阅者都可以看作是一个链表。例如：

// S{n} -> Subscriber `n`, where `n` depends on the order in which the subscribers are created

of(/* ... */)
  .pipe(
    operatorA(), // S{4}
    operatorB(), // S{3}
    operatorC(), // S{2}
  ).subscribe({ /* ... */ }) // S{1}; the observer is converted into a `Subscriber`

S{n}是 的父（目的地）S{n+1}，意思S{1}是 的目的地S{2}，S{2}是的目的地，S{3}依此类推。

StackBlitz

---

意想不到的结果

比较这些：

of(2, 1, 0)
  .pipe(
    mergeMap(v => timer(v * 100).pipe(mapTo(v)))
  ).subscribe(console.log)
// 0 1 2

of(2, 1, 0)
  .pipe(
    mergeMap(v => timer(v).pipe(mapTo(v)))
  ).subscribe(console.log)
// 1 0 2

根据MDN：

指定的时间量（或延迟）不是保证执行时间，而是最短执行时间。在主线程上的堆栈为空之前，您传递给这些函数的回调无法运行。

因此，像 setTimeout(fn, 0) 这样的代码将在堆栈为空时立即执行，而不是立即执行。如果您执行 setTimeout(fn, 0) 之类的代码，但在运行从 1 到 100 亿的循环后立即执行，您的回调将在几秒钟后执行。

MDN 的这一部分也应该澄清一些事情。

我会说这是特定于环境的，而不是特定于 RxJs 的。

在第二个片段中，延迟是连续的，这就是您得到意外结果的原因。如果您稍微增加延迟，例如：timer(v * 2)，您应该会得到预期的行为。