如何等待两个Observable各至少发射一次后返回第三个Observable的Promise？

我明白你的需求了——要等$A和$B各自至少发射一次值（并且确认它们完成），然后再返回$C转成的Promise对吧？之前的实现踩坑很正常，RxJS的操作符选不对就容易出问题，我给你分两种场景给出正确写法：

场景1：必须等$A和$B都完成（且各自至少发过一次值） #

这种场景下用forkJoin最合适，它会严格等待所有传入的Observable完成，并且要求每个Observable至少发出一次值，刚好匹配你“确认已完成”+“各至少发射一次”的需求：

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

// 等$A和$B都完成后，触发$C并转为Promise
const resultPromise: Promise<CType> = forkJoin([$A, $B])
  .pipe(
    // 完全忽略$A和$B的结果，直接切换到目标Observable $C
    switchMap(() => $C)
  )
  .toPromise();

注意事项 #

如果$A或$B存在不发出任何值就直接完成的可能，forkJoin会一直不触发（因为它要求每个Observable至少发一次值）。这时候可以给它们加上defaultIfEmpty操作符，兜底一个默认值：

import { forkJoin, defaultIfEmpty } from 'rxjs';
import { switchMap } from 'rxjs/operators';

// 给可能空值的Observable加默认值兜底
const safe$A = $A.pipe(defaultIfEmpty(null)); // 替换成适合AType的默认值
const safe$B = $B.pipe(defaultIfEmpty(null));

const resultPromise: Promise<CType> = forkJoin([safe$A, safe$B])
  .pipe(switchMap(() => $C))
  .toPromise();

场景2：只要$A和$B各发过一次值，无需等待它们完成 #

如果你的需求是只要两者都发过一次值，就立即触发$C（不管$A/$B后续是否还会发值或完成），那用combineLatest配合take(1)就可以了：

import { combineLatest } from 'rxjs';
import { switchMap, take } from 'rxjs/operators';

// 只要$A和$B各发过一次值，就立即触发$C
const resultPromise: Promise<CType> = combineLatest([$A, $B])
  .pipe(
    take(1), // 只取第一次满足条件的时机，避免后续重复触发
    switchMap(() => $C)
  )
  .toPromise();

为什么之前的实现可能出问题？ #

常见的坑包括：

• 用了merge或concat：只会等待其中一个Observable完成/发值，无法确保两者都满足条件
• 用了race：只会取第一个完成的Observable，完全不符合“两者都要满足”的需求
• 没处理空值Observable：导致forkJoin一直不触发

内容的提问来源于stack exchange，提问作者Kyle V.