在来这里之前，我已经阅读了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 个可观察量一起使用。
老实说，我很难理解这个运算符，如果有人能帮助我以简单的方式理解它，除了了解它在我之前展示的代码中的用法外，我将不胜感激。提前问候。谢谢！
mergeMap像许多其他所谓的高阶映射运算符一样，维护一个或多个内部可观察量。
使用外部值和提供的函数创建内部可观察量。外部值本质上只是从其源接收的值。例如：
of(1, 2, 3).pipe(
mergeMap((outerValue, index) => /* ... return an observable ... */)
).subscribe(); // `outerValue`: 1, 2, 3 (separately)
当外部值进入时，将创建一个新的内部可观察量。我认为理解这一点的最好方法是查看源代码：
// `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，我们稍后再看。
现在，当内部可观察量发出时会发生什么？在进一步讨论之前，值得一提的是，尽管很明显，但内部可观察量需要一个内部订阅者。我们可以在上面的_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);
}
当内部可观察对象发出时，将调用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，该 obs 将在300毫秒内发出。由于当前脚本尚未完成执行，因此尚未考虑回调队列。现在1到了，并将创建一个内部 obs，它将在100毫秒内发出。
mergeMap现在有 3 个内部可观察量，并将传递任何内部可观察量发射的内值.
正如预期的那样，我们得到1，2，3。
这就是mergeMap所做的。混合可观察量可以这样想：如果一个外部值来了，并且已经创建了一个内部可观察量，那么mergeMap只是说："没问题，我只是创建一个新的内部可观察量并订阅它"。
concurrent和buffer呢
mergeMap可以给出第二个参数，concurrent它指示应该同时处理多少个内部可观察量。这些活动的内部可观察量数量使用active属性进行跟踪。
如_next方法所示，如果active >= concurrent，outerValues将被添加到一个buffer，这是一个队列(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.
链中的下一个订阅者呢
运算符是返回另一个函数的函数，该函数接受一个可观察量作为其唯一参数并返回另一个可观察量。订阅流时，运营商返回的每个可观察量都有自己的订阅者。
所有这些订阅者都可以被视为一个链表。例如：
// 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}的目的地，依此类推。
堆栈闪电战
<小时 />意外结果
比较这些：
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)，您应该得到预期的行为。
因此，合并映射主要用于同时解析多个内部可观察量，当所有内部可观察量被解析时，外部可观察量将解析。我希望这有所帮助。
想象一下，您必须从某个异步源读取id列表，无论是远程服务，数据库还是文件系统上的文件。
假设您必须为每个id启动异步查询才能获取详细信息。
想象一下，您必须收集每个id的所有详细信息并执行其他操作。
你最终会有一个初始的可观察量发出一个列表，然后是由该列表生成的一堆可观察量。这是你会用mergeMap.
代码如下所示
mySourceObs = getIdListFromSomewhere();
myStream = mySourceObs.pipe(
// after you get the list of the ids from your service, you generate a new strem
// which emits all the values of the list via from operator
concatMap(listOfIds => from(listOfIds)),
// for each id you get the details
mergeMap(id => getDetails(id),
)
如果您订阅myStream您将获得一个详细信息数据流，原始列表的每个id一个。代码将很简单
myStream.subscribe(
detail => {
// do what you have to do with the details of an id
}
)
有关问题中引用的代码的更多信息
我对使用mergeMap的代码段的理解如下：
- 您使用
auth.getNewAccessToken获取新令牌 - 如果出现问题，请重试 3 次
- 当你收到一个新的令牌时，你会做一些事情，然后你用
next.handle(clonHttp)克隆一些东西
关键点是auth.getNewAccessToken和next.handle(clonHttp)都是返回可观察量的异步调用。
在这种情况下，您要确保首先从auth.getNewAccessToken获得响应，然后才调用next.handle(clonHttp)。
在这种情况下，编写此类逻辑的最佳方法是使用concatMap，以确保第二个 Observable 连接到第一个可观察量的成功完成。
mergeMap和switchMap也可以在这种情况下工作，因为auth.getNewAccessToken只发出一次然后完成，但正确的语义由concatMap给出(顺便说一下，这与并发设置为 1 的mergeMap相同，但这是另一回事)。