为什么订阅独立工作时我的 forkJoin 订阅未达到？答案

【问题标题】：Why is my forkJoin subscription not reached while the subscriptions work independently?为什么订阅独立工作时我的 forkJoin 订阅未达到？
【发布时间】：2021-04-09 09:46:45
【问题描述】：

我有两个可观察对象，我想将它们合并到来自rxjs 的forkJoin 方法中。独立执行 observables 可以工作，但使用 forkJoin 无法达到 pipe finalize/subscribe 方法。

我的.component.ts

....
const req1 = this.userService.getUser(this.loggedInUser.userId);
const req2 = this.boardGames$;
this.subscriptions$ = forkJoin([req1, req2])
  .pipe(
    finalize(() => {
      console.log('pipe'); // Is not reached
    })
  )
  .subscribe(([obj1, obj2]) => {
    console.log('subscribe'); // Is not reached
  }, err => console.log(err), ()=>console.log('compl'));
req1.subscribe((aa) => console.log(aa)); // This is logged
req2.subscribe((bb) => console.log(bb)); // This is logged
....

我正在使用Angularfire2 进行请求。我不确定这是否会成为问题，因为订阅独立工作。 import { AngularFirestore } from 'angularfire2/firestore';

我在这里缺少什么？

【问题讨论】：

这是因为我想取消订阅组件的销毁。不幸的是，这没有帮助
这也没有打印任何东西
Stackblitz 就像一个在线编辑器，您可以通过在其中编写代码然后共享 url 来复制问题。 Stackblitz
你的一个源 Observables req1 或 req2 永远不会完成，所以 forkJoin() 不会发出任何东西
@martin 事件虽然最后两行有效？

标签： angular rxjs observable subscription fork-join

【解决方案1】：

forkjoin() 需要您的两个订阅都完成才能真正加入。因此，如果您的任何一个订阅都没有完成，那么forkjoin() 将永远无法到达。如果您使用的是 firebase，则您的 observables 不会完成。

如果您的订阅未完成并且您需要来自两个 observable 的流，那么您应该尝试使用 combineLatest()。这需要两个活动订阅，一旦每个订阅发出一个值，就会将这些值加入一个订阅中，并继续发出值直到完成。

Here is a link for combineLatest

如果您只需要在调用 firebase 之前检查用户是否有效，请尝试switchMap()。这会将您的用户 observable 切换到您的棋盘游戏 observable，而您将只处理棋盘游戏 observable。

【讨论】：

【解决方案2】：

forkJoin 仅在所有可观察对象完成时才发出。我看不到您的其余代码（例如 boardGames$ observable 是什么）。您很可能正在使用在第一次发射后无法完成的 observable，这是 AngularFirestore 的预期行为，因为您最常订阅数据库 (Firebase) 中的更改。

如果您需要在某些 observable 发出时获取最新值，请使用 combineLatest。请记住，只有在每个源 observables 发出时，它才会开始发出。

combineLatest([
    this.userService.getUser(this.loggedInUser.userId), 
    this.boardGames$
]).subscribe(([user, boardGames]) => {
     // Don't forget to unsubscribe
});

使用merge ir 你想将可观察对象合并为一个可观察对象。它适用于您当前的情况。像这样：

merge(
  this.userService.getUser(this.loggedInUser.userId).pipe(map(entity => ({entity, type: 'user'}))),
  this.boardGames$.pipe(map(entity => ({entity, type: 'boardGames'})))
).subscribe(({entity, type}) => {
    // Don't forget to unsubscribe
})

使用forkJoin，您可以这样实现：

const req1 = this.userService.getUser(this.loggedInUser.userId).pipe(take(1));
const req2 = this.boardGames$.pipe(take(1));
this.subscriptions$ = forkJoin([req1, req2]).subscribe(() => {
    // I will complete after both observables emits.
});

请注意，即使使用 take(1)，您仍然需要处理订阅，因为如果某些 observable 永远不会发出并且组件被破坏，您将有内存泄漏。有awesome library 用于处理没有样板的订阅。

【讨论】：