Angular中订阅Observable时如何避免竞态条件？附代码求助

原代码的核心问题 #

你的第一段代码里，subscribe是异步操作，this.store.dispatch会在assigneeId$和customer$发射值之前执行，导致userId和customerId大概率是undefined，同时两个Observable的发射顺序不可控，确实存在竞态条件。

嵌套订阅不生效的常见原因 #

你写的嵌套订阅没起作用，大概率是这两个原因之一：

• assigneeId$从未发射过值，外层订阅根本没触发
• customer$在assigneeId$发射后没有新值发射（比如它是只发射一次的冷Observable，或者之前已经发射过值但没有重复发射）

另外嵌套订阅本身是不推荐的写法，会导致内存泄漏、逻辑混乱，一旦Observable多次发射，还会触发大量不必要的dispatch。

正确的RxJS组合写法 #

根据你的业务场景，推荐用以下几种RxJS操作符来组合Observable，彻底消除竞态：

场景1：需要两个Observable都至少发射一次，之后任意一个发射新值都触发dispatch

用combineLatest，它会在所有源Observable都发射过至少一次值后，每当其中一个发射新值时，就把所有Observable的最新值组合起来：

onUpdate(relation: RelationTypes) {
  const combined$ = combineLatest([this.assigneeId$, this.customer$]);
  
  this.subscription.add(
    combined$.subscribe(([userId, customer]) => {
      this.store.dispatch(actions.changeUserRelation({
        payload: {
          userId,
          customerId: customer.email,
          relation
        }
      }));
    })
  );
}

场景2：仅当assigneeId$发射新值时，才取customer$的最新值触发dispatch

如果customer$是持续存在的状态（比如从Store里获取的），用withLatestFrom更合适，它只会在源Observable（这里是assigneeId$）发射时，组合最新的其他Observable值：

onUpdate(relation: RelationTypes) {
  this.subscription.add(
    this.assigneeId$
      .pipe(withLatestFrom(this.customer$))
      .subscribe(([userId, customer]) => {
        this.store.dispatch(actions.changeUserRelation({
          payload: {
            userId,
            customerId: customer.email,
            relation
          }
        }));
      })
  );
}

场景3：两个Observable都只会发射一次（比如HTTP请求）

用forkJoin，它会等待所有Observable完成后，返回最后一次发射的值：

onUpdate(relation: RelationTypes) {
  const joined$ = forkJoin([this.assigneeId$, this.customer$]);
  
  this.subscription.add(
    joined$.subscribe(([userId, customer]) => {
      this.store.dispatch(actions.changeUserRelation({
        payload: {
          userId,
          customerId: customer.email,
          relation
        }
      }));
    })
  );
}

额外注意事项 #

• 别忘了在组件销毁时取消所有订阅，避免内存泄漏，比如可以定义一个destroy$ = new Subject<void>()，然后在每个订阅里加pipe(takeUntil(this.destroy$))，最后在ngOnDestroy里调用this.destroy$.next(); this.destroy$.complete();
• 如果customer$可能为空，建议加filter操作符确保拿到有效值，比如pipe(filter(customer => !!customer))

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