如何在已有独立API订阅逻辑中使用RxJS的forkJoin？

错误原因 #

你遇到的报错是因为传给forkJoin的不是Observable（可观察对象），而是Subscription（订阅对象）。原来的getFirstApi和getSecondApi函数里直接调用了.subscribe()，这个方法返回的是订阅实例，不是原Observable，而forkJoin只接受Observable、Promise这类可异步等待的对象。

正确实现方式 #

方案一：让API函数返回Observable，统一在forkJoin中订阅

修改组件中的函数，让它们返回服务返回的Observable，不在函数内部订阅，而是把订阅和数据处理、后续操作统一放到forkJoin中：

ngOnInit() {
  // 获取两个API的Observable
  const firstApi$ = this.getFirstApi();
  const secondApi$ = this.getSecondApi();

  // 用forkJoin等待两个请求都完成
  forkJoin([firstApi$, secondApi$]).subscribe(([firstData, secondData]) => {
    // 先处理各自的数据
    this.doingSomethingWithData(firstData);
    this.doingSomethingWithData(secondData);
    // 执行后续操作
    this.executeRestCodeNow();
  });
}

// 返回Observable，不在这里订阅
getFirstApi(): Observable<any> {
  return this.myService.firstApi();
}

// 注意：原代码里这里误写为firstApi()，要改成secondApi()
getSecondApi(): Observable<any> {
  return this.myService.secondApi();
}

doingSomethingWithData(data: any) {
  // 你的数据处理逻辑
}

executeRestCodeNow() {
  // 两个请求完成后要执行的逻辑
}

方案二：保留函数内的数据处理逻辑，用tap操作符

如果不想改变原有函数的职责（在函数内处理数据），可以用RxJS的tap操作符在Observable流中处理数据，同时返回原Observable给forkJoin：

ngOnInit() {
  const firstApi$ = this.getFirstApi();
  const secondApi$ = this.getSecondApi();

  forkJoin([firstApi$, secondApi$]).subscribe(() => {
    // 两个请求都完成后执行后续操作
    this.executeRestCodeNow();
  });
}

getFirstApi(): Observable<any> {
  return this.myService.firstApi().pipe(
    // tap操作符用于在流中执行副作用（比如处理数据），不改变流本身
    tap(data => this.doingSomethingWithData(data))
  );
}

// 修正原代码的笔误，调用secondApi()
getSecondApi(): Observable<any> {
  return this.myService.secondApi().pipe(
    tap(data => this.doingSomethingWithData(data))
  );
}

doingSomethingWithData(data: any) {
  // 你的数据处理逻辑
}

executeRestCodeNow() {
  // 后续操作逻辑
}

额外注意 #

原代码中getSecondApi函数里调用的是this.myService.firstApi()，这是笔误，必须改成this.myService.secondApi()，否则会重复请求第一个接口，达不到并行请求两个接口的目的。

内容的提问来源于stack exchange，提问作者Snk