Angular forkJoin 订阅未触发答案

【问题标题】：Angular forkJoin Subscribe not firingAngular forkJoin 订阅未触发
【发布时间】：2021-04-24 06:08:25
【问题描述】：

我有几种方法

    getFirstDataBunch() {
        return this.repository.getData(this.parameter).pipe(
            switchMap(
                result => {
                    //Do something with result;
                    return of(true);
                }
            )
        );
    }

然后我打电话给我的ngOnInit：

    function ngOnInit(){
        forkJoin(this.getFirstDataBunch(), this.getSecondDataBunch(), this.getThirdDataBunch()).subscribe(
            results => {
                //do stuff with results; 
            }
        );
    }

我可以看到所有的 observable 都被调用了，它们返回一个 observable，但是 forkJoin 订阅方法从未被调用，从我读到的 forkJoin 仅在所有 observable 返回并完成时才会触发，而且我确实相信是这样，那为什么不开除呢？

我还看到有些人只是选择combineLatest，但因为这些方法只会在我真的不需要继续寻找更新时才返回。

【问题讨论】：

显然，至少有一个dataBunch() 流没有完成。你检查了吗？
@mbojko 我该怎么做？函数与您在此处看到的完全一样，我对一些数据进行 GET 请求，进行一些转换，然后返回一个 observable of(true)，这样 switchMap 确实有一个返回类型 Observable。

标签： angular rxjs observable

【解决方案1】：

我会简单地使用您想要/需要的组合运算符并将其与take(1) 一起传递。

为什么是take() 运算符？

因为 take 返回一个 Observable，它只发出源 Observable 发出的前 X 个值。我们设置了多少排放量。

例子：

import { of } from 'rxjs';
import { take } from 'rxjs/operators';

//emit 1,2,3,4,5
const source = of(1, 2, 3, 4, 5);
//take the first emitted value then complete
const example = source.pipe(take(1));
//output: 1
const subscribe = example.subscribe(console.log);

//output: 1, 2, 3
const example2 = source.pipe(take(3)).subscribe(console.log);

如果你想 100% 确保避免任何可能的内存泄漏，我建议也使用 takeUntil()（因为 take(1) 可能由于某种原因永远卡住，如果 observable 永远不会发出）

.pipe(take(1), takeUntil(this.destroy$))

作为一般建议，始终将 takeUntil() 作为管道的最后一个参数。

关于什么组合运算符：

forkJoin --> 如您所写，等待所有可观察对象完成，然后从每个对象发出最后一个发出的值（如Promise.all）。它会起作用，但每个可观察对象都需要完成（不仅仅是发射！）。

zip --> 等待所有可观察对象发出，然后将值作为数组发出（按顺序 - 下图）。我可能会用这个。

combineLatest --> 等待所有 observables 发射，发射一次。然后，每当任何输入的 Observable 发出一个值时，它都会发出发出的 observable 的新值以及其他（未发出的）最后一次发出的值。当然，这也适用于 take(1)...

【讨论】：

【解决方案2】：

您的方法未完成，您应该通过first() 运算符或take(1) 手动完成。

function ngOnInit(){
        forkJoin(
         this.getFirstDataBunch().pipe(first()), 
         this.getSecondDataBunch().pipe(first()), 
         this.getThirdDataBunch().pipe(first())
        ).subscribe(
            results => {
                //do stuff with results; 
            }
        );
    }

【讨论】：