如何在RxJS中待两个观察者获取结果后执行关联方法？

方法一：使用RxJS操作符（推荐，声明式） #

这种方式利用RxJS的管道操作符，将各自的业务逻辑与时机监听分离，保持代码的响应式和整洁。

步骤1：封装各自的处理逻辑 #

将原订阅中的赋值逻辑通过tap操作符封装到对应的流管道中，tap用于执行副作用（比如给类属性赋值）且不改变流的输出：

// 处理specialties流的逻辑
const processedSpecialties$ = specialties$.pipe(
  tap(value => {
    this.specialtiesOptions = value['result'].map(specialty => ({
      ...specialty,
      selected: false
    }));
  })
);

// 处理partner流的逻辑
const processedPartner$ = partner$.pipe(
  tap((value: Partner) => {
    // 保留原有的其他业务逻辑
    this.partner = mutate(value);
  })
);

步骤2：合并流并触发共享方法 #

根据你的流类型选择合适的合并操作符：

场景A：两个流都是单次发射（比如HTTP请求，发出值后立即完成）

使用forkJoin，它会等待所有流完成后才发出合并结果：

import { forkJoin } from 'rxjs';

forkJoin([processedSpecialties$, processedPartner$]).subscribe(() => {
  // 调用你的共享方法
  this.yourSharedMethod();
});

场景B：流可能多次发射，但只需要首次两个都拿到值后调用一次方法

使用combineLatest配合take(1)，它会在两个流都至少发出一次值后触发，take(1)确保只执行一次：

import { combineLatest, take } from 'rxjs';

combineLatest([processedSpecialties$, processedPartner$]).pipe(
  take(1)
).subscribe(() => {
  this.yourSharedMethod();
});

这种方式下，各自的业务逻辑完全分离在对应的流管道中，合并操作只负责监听时机，符合你的需求。

方法二：命令式检查（备选） #

如果你更倾向于命令式写法，可以通过标志位来追踪两个流的完成状态：

private isSpecialtiesReady = false;
private isPartnerReady = false;

// 原specialties订阅逻辑
specialties$.subscribe(value => {
  this.specialtiesOptions = value['result'].map(specialty => ({
    ...specialty,
    selected: false
  }));
  this.isSpecialtiesReady = true;
  this.checkAndExecute();
});

// 原partner订阅逻辑
partner$.subscribe({
  next: (value: Partner) => {
    // 保留原有业务逻辑
    this.partner = mutate(value);
    this.isPartnerReady = true;
    this.checkAndExecute();
  }
});

// 检查并调用共享方法
private checkAndExecute() {
  if (this.isSpecialtiesReady && this.isPartnerReady) {
    this.yourSharedMethod();
    // 若只需调用一次，可重置标志位或取消订阅
    // this.isSpecialtiesReady = false;
    // this.isPartnerReady = false;
  }
}

这种方式简单直接，但属于命令式编程，不如RxJS的声明式方式优雅，且需要手动处理重复触发的问题。

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