Angular 6 : How to make a set of service calls in parallel and process Data once all of them are completed

Question

我正在使用 rxjs 6+ 开发 Angular 6+ 网络应用程序。

有一组服务(都是不同的，进行不同的 http 调用)。这些服务基本上用于初始化应用程序。

我的要求是并行调用这些服务并等待所有服务完成。完成后，我需要处理收到的回复。我对此进行了很多研究，发现 forkJoin 和 combineLatest 可以成为我的 friend 。我已经通过以下方式实现了相同的方式:

 getInitialData() {

 this._authService.openCorsConnection().subscribe(res => { 
  forkJoin(
    this._authService.x(),
    this._tService.getSystemDate(),
    this._tService.getTemp(),
    this._tService.getSettings()
  ).pipe(
    catchError( err => {
      console.log('gor error',err);
      if (err.status == 401) {
          // this._router.navigateByUrl('/unauthorize');
          return EMPTY;
      } else {
          return throwError(err);
      }
 }),
    map(([x,y,z,p])=>{
           console.log('sucesss',{x,y,z,p});          
            return {x,y,z,p}
    }),
  finalize(()=>{
        console.log('Executing Finally');
        processData();

  })
  );
});
 }

但是这些都没有执行。以下是服务:

x(): Observable<any> {          
const xUrl  = EnvironmentConfiguration.endPointConfig.xServiceEndpoint;
          let serviceMethod: String = 'x';
          let requestObj: any = this._utilHelperSvc.getRequestObject(xUrl  , { "y": "" }, serviceMethod);
          return this._http.post<any>(requestObj.url,{ "pid": "" } , requestObj)
          .pipe(
            catchError(this.handleError('x', {}))
          );
      }

我需要修改服务吗？我不知道如何解决这个问题。

任何人都可以为此提出一些更好的解决方案或不同的方法吗？

谢谢。

Answer 1

长话短说

使用zip或 forkjoin .

演示:

• Demo with forkJoin
• Demo with zip

解释

您不需要等待 forkJoin 来处理响应，它以相反的方式工作。相反，准备所有需要完成的工作，然后等待完成。

我的意思是:

let process1$ = timer(1000);
let process2$ = timer(2000);
let process3$ = timer(3000);

const doSthg1 = pipe(
  tap(() => console.log('Process 1 has finished')),
  map(() => 'Process 1 has finished')
)

const doSthg2 = pipe(
  tap(() => console.log('Process 2 has finished')),
  map(() => 'Process 2 has finished')
)

const doSthg3 = pipe(
  tap(() => console.log('Process 3 has finished')),
  map(() => 'Process 3 has finished')
)

forkJoin(doSthg1(process1$), doSthg2(process2$), doSthg3(process3$)).subscribe(() => {
  console.log('Now I am complete');
});

Demo with forkJoin

只要您的流程没有链接，这就有效，即一个输入不依赖于另一个的输出。

为什么我的代码不起作用？

因为您实际上并没有订阅 forkJoin。解决方案:例如，您可以使用 concatMap 将 this._authService.openCorsConnection() 转换为另一个 Observable:

getInitialData(){
  this._authService.openCorsConnection().pipe(
    concatMap(() => forkJoin(this._authService.x(),
      this._tService.getSystemDate(),
      this._tService.getTemp(),
      this._tService.getSettings()
    )),
    catchError(err => {
      console.log('gor error', err);
      if (err.status == 401) {
        return EMPTY;
      } else {
        return throwError(err);
      }
    }),
  ).subscribe(() => processData());
}

压缩

如果满足以下条件，您可以使用 zip 而不是 forkJoin:

1. 所有进程只发出一次，然后完成:例如 http Get 或 Post 请求。
2. 您想从您的来源获得联合输出。

实现:

zip(doSthg1(process1$), doSthg2(process2$), doSthg3(process3$))
.subscribe(
      ([x,y,z]) => {
      console.log(`x=${x}, y=${y}, z=${z}`);
});

Demo with zip