问题描述
我很难理解 rxJs 中 flatMap
和 concatMap
之间的区别。
我能理解的最明确的答案是这里difference-between-concatmap-and-flatmap
所以我自己去尝试了。
import "./styles.css";
import { switchMap,flatMap,concatMap } from "rxjs/operators";
import { fromFetch } from "rxjs/fetch";
import { Observable } from "rxjs";
function createObs1() {
return new Observable<number>((subscriber) => {
setTimeout(() => {
subscriber.next(1);
subscriber.complete();
},900);
});
}
function createObs2() {
return new Observable<number>((subscriber) => {
setTimeout(() => {
subscriber.next(2);
//subscriber.next(22);
//subscriber.next(222);
subscriber.complete();
},800);
});
}
function createObs3() {
return new Observable<number>((subscriber) => {
setTimeout(() => {
subscriber.next(3);
//subscriber.next(33);
//subscriber.next(333);
subscriber.complete();
},700);
});
}
function createObs4() {
return new Observable<number>((subscriber) => {
setTimeout(() => {
subscriber.next(4);
subscriber.complete();
},600);
});
}
function createObs5() {
return new Observable<number>((subscriber) => {
setTimeout(() => {
subscriber.next(5);
subscriber.complete();
},500);
});
}
createObs1()
.pipe(
flatMap((resp) => {
console.log(resp);
return createObs2();
}),flatMap((resp) => {
console.log(resp);
return createObs3();
}),flatMap((resp) => {
console.log(resp);
return createObs4();
}),flatMap((resp) => {
console.log(resp);
return createObs5();
})
)
.subscribe((resp) => console.log(resp));
console.log("hellooo");
我在这里用过那个游乐场playground example
问题
1) 根据我的理解,flatMap 的使用应该混合输出,以便控制台日志类似于 (1,3,2,4,5)。我已经尝试了 30 多次并且总是出现在同一行 (1,5)
我做错了什么或有什么不可告人的错误?
2)
如果在 createObs2()
和 createObs3()
上删除注释并将代码包含在多个发出的事件中,那么事情就会变得一团糟。即使您更改为 concatMap 它也会使事情变得混乱并且结果喜忧参半。我期望只出现一次的多个数字多次出现。结果可以是 (1,33,22,5,5) 为什么会这样?
我如何在操场上测试示例。我只是从最后一个 console.log("hello") 中删除了 1 个字母。只有一个更改,例如 console.log("heloo") 然后观察到并再次编译项目并在控制台中打印输出。
编辑:我使用 flatMap 和 concatMap 的原因是为了使用 http 库在 angular 中找到嵌套订阅的替代品。
createObs1().subscribe( (resp1) => {
console.log(resp1);
createObs2().subscribe( (resp2) => {
console.log(resp2);
createObs3().subscribe( (resp3) => {
console.log(resp3);
createObs4().subscribe( (resp4) => {
console.log(resp4);
createObs5().subscribe( (resp5) => {
console.log(resp5);
})
})
})
})
})
解决方法
您的测试场景并不足以看出这两个运算符之间的差异。在您的测试用例中,每个 observable 只发出 1 次。如果 observable 只发出一个值,那么 concatMap
和 flatMap
(又名 mergeMap)之间真的没有什么不同。只有在多次排放时才能看到差异。
所以,让我们使用不同的场景。让我们有一个 source$
observable,它只是每 1 秒发出一个递增的整数。然后,在我们的“Higher Order Mapping Operator”(concatMap
& mergeMap
)中,我们将返回一个每 1 秒发出可变次数的 observable,然后完成。
// emit number every second
const source$ = interval(1000).pipe(map(n => n+1));
// helper to return observable that emits the provided number of times
function inner$(max: number,description: string): Observable<string> {
return interval(1000).pipe(
map(n => `[${description}: inner source ${max}] ${n+1}/${max}`),take(max),);
}
然后让我们基于 source$
和 inner$
定义两个单独的 observable;一种使用 concatMap
,一种使用 flatMap
并观察输出。
const flatMap$ = source$.pipe(
flatMap(n => inner$(n,'flatMap$'))
);
const concatMap$ = source$.pipe(
concatMap(n => inner$(n,'concatMap$'))
);
在查看输出的差异之前,让我们先谈谈这些运算符的共同点。他们俩:
- 订阅传入函数返回的observable
- 从这个“内部可观察对象”中发出排放
- 取消订阅内部 observable
不同之处在于它们如何创建和管理内部订阅:
concatMap
- 一次只允许一个内部订阅。当它接收到排放时,它一次只会订阅一个内部可观察对象。所以它最初会订阅“emission 1”创建的observable,只有在它完成后才会订阅“emission 2”创建的observable。这与 concat
静态方法的行为方式一致。
flatMap
(aka mergeMap
) - 允许许多内部订阅。因此,当接收到新的排放时,它将订阅内部可观察量。这意味着排放不会按任何特定顺序进行,因为它会在其任何内部可观察对象发出时发出。这与 merge
静态方法的行为方式一致(这就是我个人更喜欢名称“mergeMap”的原因)。
这是一个 StackBlitz,它显示了上述 observables concatMap$
和 mergeMap$
的输出:
希望以上解释能帮助您解决问题!
#1 - “使用 flatMap 应该混合输出”
这没有像您预期的那样工作的原因是因为只有一个发射通过 flatMap
,这意味着您只有一个“内部可观察”发射值。如上例所示,一旦 flatMap 接收到多个发射,它就可以拥有多个独立发射的内部 observable。
#2 - “...并包含带有多个发出事件的代码,然后事情就会变得一团糟。”
“事情变得一团糟”是因为有多个发出值的内部订阅。
对于您提到的关于使用 concatMap
并且仍然获得“混合”输出的部分,我不希望这样。当启用“自动保存”时,我在 StackBlitz 中看到了可观察到的异常行为(似乎有时它不会完全刷新,并且旧订阅似乎在自动刷新后仍然存在,这会导致控制台输出非常混乱 )。也许代码沙箱也有类似的问题。
#3 - “我使用 flatMap 和 concatMap 的原因是为了使用 http 库找到嵌套订阅的替代品”
这是有道理的。您不想弄乱嵌套订阅,因为没有一个很好的方法来保证内部订阅会被清理。
在大多数使用 http 调用的情况下,我发现 switchMap
是理想的选择,因为它会减少您不再关心的内部 observable 的排放。假设您有一个从路由参数中读取 id
的组件。它使用此 id
进行 http 调用以获取数据。
itemId$ = this.activeRoute.params.pipe(
map(params => params['id']),distinctUntilChanged()
);
item$ = this.itemId$.pipe(
switchMap(id => http.get(`${serverUrl}/items/${id}`)),map(response => response.data)
);
我们希望 item$
仅发出“当前项目”(对应于 url 中的 id)。假设我们的 UI 有一个按钮,用户可以点击以通过 id
导航到下一个项目,并且您的应用发现自己有一个喜欢点击的用户,他一直在粉碎该按钮,这比 http 调用更快地更改了 url 参数可以返回数据。
如果我们选择 mergeMap
,我们最终会得到许多内部 observable,它们会发出所有这些 http 调用的结果。充其量,屏幕会随着所有这些不同的呼叫返回而闪烁。在最坏的情况下(如果调用出现乱序),UI 将显示与 url 中的 id 不同步的数据:-(
如果我们选择 concatMap
,用户将被迫等待所有的 http 调用串行完成,即使我们只关心最近的那个。
但是,对于 switchMap
,每当接收到新的发射 (itemId
) 时,它将取消订阅前一个内部可观察对象并订阅新的。这意味着它永远不会发出不再相关的旧 http 调用的结果。 :-)
需要注意的一点是,由于 http observables 只发出一次,各种操作符(switchMap
、mergeMap
、concatMap
)之间的选择似乎没有什么区别,因为它们都为我们执行“内部可观察处理”。但是,最好让您的代码面向未来,并选择真正为您提供所需行为的代码,如果您开始接收多个发射。
每次第一个 observable 发出时,都会在 flatMap
中创建第二个 observable 并开始发出。但是,第一个 observable 的值不会再继续传递。
每次第二个 observable 发出时,下一个 flatMap
都会创建第三个 observable,依此类推。同样,进入 flatMap
的原始值不会进一步传递。
createObs1()
.pipe(
flatMap(() => createObs2()),// Merge this stream every time prev observable emits
flatMap(() => createObs3()),// Merge this stream every time prev observable emits
flatMap(() => createObs4()),// Merge this stream every time prev observable emits
flatMap(() => createObs5()),// Merge this stream every time prev observable emits
)
.subscribe((resp) => console.log(resp));
// OUTPUT:
// 5
因此,只有从 createObs5()
发出的值才真正发送给观察者。先前 observable 发出的值刚刚触发了新 observable 的创建。
如果您使用 merge
,那么您会得到您所期望的:
createObs1()
.pipe(
merge(createObs2()),merge(createObs3()),merge(createObs4()),merge(createObs5()),)
.subscribe((resp) => console.log(resp));
// OUTPUT:
// 5
// 4
// 3
// 2
// 1