如何合并两个Observable并根据最后发射的Observable执行不同函数？

你可以通过给每个Observable的发射值打上来源标记，结合merge和scan操作符来实现需求，代码更简洁直观：

步骤1：给Observable添加来源标识 #

先把两个流的每个发射值包装成包含来源标识和实际值的对象：

const labeledObs1 = this.obs1.pipe(map(val => ({ source: 'obs1', val })));
const labeledObs2 = this.obs2.pipe(map(val => ({ source: 'obs2', val })));

步骤2：合并流并维护状态 #

用merge合并两个带标记的流，再用scan操作符维护两个流的最新值，同时记录最后发射的来源：

this.desiredObservable = merge(labeledObs1, labeledObs2).pipe(
  scan((state, curr) => ({
    // 更新对应来源的最新值
    obs1: curr.source === 'obs1' ? curr.val : state.obs1,
    obs2: curr.source === 'obs2' ? curr.val : state.obs2,
    // 记录最后发射的来源
    lastSource: curr.source
  }), { 
    // 初始状态：obs1用BehaviorSubject的当前值，obs2设为初始默认值（根据实际情况调整）
    obs1: this.obs1.value, 
    obs2: '' 
  }),
  // 根据最后发射来源调用对应函数
  map(state => 
    state.lastSource === 'obs1'
      ? func1(state.obs1, state.obs2)
      : func2(state.obs1, state.obs2)
  )
);

方案优势 #

• 只用RxJS基础操作符：map、merge、scan都是常用操作符，无需引入复杂工具函数
• 逻辑直接：通过标记来源就能明确追踪最后发射的Observable，避免额外的状态判断
• 适配BehaviorSubject：初始状态直接复用obs1的当前值，符合BehaviorSubject的特性

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