如何让RxJS combineLatest在DAG同源路径下仅发射一次？

需求说明 #

我们有如下依赖关系的Observable：

A C
  /| |
 B | |
  \|/
   D

• B由A派生，订阅A的发射
• D同时订阅A、B、C三个Observable
• 期望的触发逻辑：
  • 当A发射新值时，D仅在A和B都完成当前值的发射后触发一次
  • 当C发射新值时，D立即触发发射

默认使用combineLatest({A, B, C})会出现问题：A更新时，D会触发两次——一次是A更新瞬间（此时B还未响应A的新值），一次是B完成A新值的处理后再次发射，导致D重复输出。

最优实现方案 #

核心思路是将存在依赖关系的A和B打包成一个单一的Observable流，消除组合时的重复触发源，再和C进行组合。

基础场景代码实现 #

如果B是A的简单同步派生（比如map转换），可以直接在A的流中同时生成B的值：

const A = new Rx.BehaviorSubject(1);
const C = new Rx.BehaviorSubject(3);

// 把A和它派生的B合并为一个流，确保A的每个值只对应一个B的值
const AWithB = A.pipe(
  map(aValue => ({
    A: aValue,
    B: aValue + 1 // 直接复用B的转换逻辑
  }))
);

// 将合并后的流与C做combineLatest
const D = Rx.combineLatest([AWithB, C]).pipe(
  map(([abPair, cValue]) => ({
    ...abPair,
    C: cValue
  }))
);

D.subscribe(console.log); 
// 初始输出：{A: 1, B: 2, C: 3}
A.next(2); // 仅输出一次：{A: 2, B: 3, C: 3}
C.next(4); // 输出：{A: 2, B: 3, C: 4}

复杂场景代码实现 #

如果B包含异步操作（比如switchMap+延迟请求），可以用switchMap确保A的每个新值都对应B处理后的结果：

const A = new Rx.BehaviorSubject(1);
const C = new Rx.BehaviorSubject(3);
// 假设B是包含异步操作的派生流
const B = A.pipe(
  switchMap(a => Rx.of(a + 1).pipe(delay(100)))
);

// 打包A和B，确保每个A的新值对应B处理后的最终结果
const AWithB = A.pipe(
  switchMap(a => B.pipe(map(b => ({A: a, B: b}))))
);

const D = Rx.combineLatest([AWithB, C]).pipe(
  map(([ab, c]) => ({...ab, C: c}))
);

方案优势 #

• 从根源上避免了A和B作为独立流带来的重复触发问题，A的每个更新只会在AWithB流中产生一次发射
• 完整保留了C更新时D立即触发的特性，完全符合需求
• 代码结构清晰，明确体现了A和B的依赖关系，可读性高

内容的提问来源于stack exchange，提问作者Adam B.