等待多个观察者完成

Question

我正在开发一个 Angular7 项目，该项目显示来自旧的、缓慢的后端的多个端点的信息。

• 我想在每个端点到达时处理和显示信息。这样用户就不必等待最慢的端点并且处理负载保持分散。

• 当最慢的端点最终返回时，组合集应集中处理。

• 不保证到达顺序。

想象的例子：

-我有 Rectangles 和 Squares 这两个数据源。

-计算每个项目的面积是一项繁重的操作。

-数据是完整的，而不是逐项的。

-我想在最后一组到达时对整组进行排序。

    Rectangles: [ {x:1 , y:3} , {x:3,y:2}   ] 30 seconds to get this
    Squares   : [ { side:10 } , { side: 2} ]  2 minutes to get this

Desired output:
    Displayed at 30 seconds: Rect-Area-3 , Rect-Area-6

    Displayed at 2 minutes: Rect-Area-3 , Square-Area-4 , Rect-Area-6 , Square-Area-100

当前代码

现在我对每个数据源都有这样的代码。来自每个来源的数据具有完全不同的 json 结构，它们几乎没有共同点。

public getDemographicData(){
        const myObserver = {
            next: serverData => {
                //
                //Process result , save.
                //Let another component handle data presentation when 
                //change detection occurs
            },
            error: err => console.error('Observer got an error: ' + JSON.stringify(err)),
            complete: () => {}
        };

        // getDemographicsData returns an observable using http.get
        this.dataService.getDemographicsData(surveyId).subscribe(myObserver);
    }

我不认为 forkjoin 是我想要的，它可以从多个 observable 中创建一个 observable。我不想等到所有数据都进来后再处理。

感谢任何帮助或 cmets，我无法在其他地方找到令人满意或可以理解的答案。我只在万不得已的情况下才在帮助网站上发帖。

Answer 1

你确实想要 forkJoin。

在每个流的tap 内部进行各个流的数据处理，并在 forkJoin 中触发组合计算，然后订阅两者。

let fastObs = of('Fast').pipe(delay(1000), tap(console.log)); // One of the faster http requests, console.log will trigger after 1 second
let slowObs = of('Slow').pipe(delay(10000), tap(console.log)); // Slower, triggers after 10 seconds

forkJoin(fastObs, slowObs).subscribe(console.log); // Triggers after 10 seconds when you have both streams completed

https://stackblitz.com/edit/typescript-qrephm?file=index.ts