RxJS多Observable并行执行 全部完成后触发任务的实现方法

你不需要二选一两套写法的优势，通过RxJS原生操作符就可以同时满足「单任务完成即时更新状态」和「所有任务完成后触发最终方法」两个需求，以下是两种常用实现：

方案1：tap 配合 forkJoin（优先推荐） #

这是最简洁、边界处理最完善的实现方式，核心是利用tap操作符的副作用能力：tap不会改变Observable原本的数据流，只会在数据抵达时执行你定义的副作用逻辑，刚好用来在单个任务返回结果时立刻更新完成标记，同时保留forkJoin等待所有任务完成的能力。
示例代码：

import { forkJoin } from 'rxjs';
import { tap } from 'rxjs/operators';

forkJoin([
  // 每个任务流单独挂载tap副作用，单任务完成立刻执行
  this.job1().pipe(
    tap(res1 => {
      this.job1Completed = true;
      // 此处可直接使用res1做单任务完成后的其他即时操作
    })
  ),
  this.job2().pipe(
    tap(res2 => {
      this.job2Completed = true;
    })
  ),
  this.job3().pipe(
    tap(res3 => {
      this.job3Completed = true;
    })
  )
]).subscribe({
  next: ([res1, res2, res3]) => {
    // 所有任务全部完成后才会执行此处
    this.finalJob();
  },
  error: (err) => {
    // 统一处理任务报错逻辑
    console.error('任务执行异常', err);
  }
});

这个方案的优势：

• 三个任务保持并行执行，性能和原生forkJoin完全一致
• 单个任务完成后会立刻触发对应tap里的状态更新，不需要等待其他任务
• 不需要手动维护完成计数，天然保证finalJob()的执行时机
• 错误处理逻辑统一，任意任务报错都会进入error回调，不会出现最终方法永远不触发的问题

方案2：手动维护完成计数（适配特殊灵活场景） #

如果你有特殊需求不想使用forkJoin，也可以自己维护完成计数器，在每个任务的订阅回调里更新计数，计数达标时触发最终方法：

// 初始化计数器
let completedNum = 0;
const totalTaskNum = 3;

// 抽离单任务完成的公共逻辑
const checkAllCompleted = () => {
  completedNum++;
  if (completedNum === totalTaskNum) {
    this.finalJob();
  }
};

// 分别订阅每个任务
this.job1().subscribe({
  next: (res1) => {
    this.job1Completed = true;
    checkAllCompleted();
  },
  error: (err) => { /* 单独处理job1异常 */ }
});

this.job2().subscribe({
  next: (res2) => {
    this.job2Completed = true;
    checkAllCompleted();
  },
  error: (err) => { /* 单独处理job2异常 */ }
});

this.job3().subscribe({
  next: (res3) => {
    this.job3Completed = true;
    checkAllCompleted();
  },
  error: (err) => { /* 单独处理job3异常 */ }
});

注意这个方案需要你手动处理每个任务的异常、重试等边界情况，否则很容易出现计数不准导致最终方法不触发/重复触发的问题，非必要场景优先选方案1。

注：你给出的分别订阅示例代码里存在笔误，jodb1Completed/jodb2Completed/jodb3Completed应为job1Completed/job2Completed/job3Completed，实际编写时需要注意变量名一致。

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