如何合并两个RxJS Observables？其中一个需通过first操作符仅触发一次

嘿，完全不用道歉！RxJS的流组合确实容易让人困惑，尤其是刚接触的时候😉。针对你的需求，我们完全可以把这两个Observable合并成一个，同时保证this.three()只执行一次，this.one()和this.two()每次queryParams变化都触发。下面提供两种可行方案：

方案一：使用scan跟踪执行状态 #

这种方案在同一个流内处理所有逻辑，通过scan操作符记录是否已经执行过this.three()：

this.route.queryParams.pipe(
  takeUntil(this.destroy$),
  scan((hasRunThree, queryParams) => {
    // 每次queryParams变化都执行one和two
    this.one();
    this.two();
    
    // 仅第一次触发时执行three，并标记状态为已执行
    if (!hasRunThree) {
      this.three();
      return true;
    }
    return hasRunThree;
  }, false) // 初始状态：未执行过three
).subscribe();

逻辑说明： #

• scan操作符会维护一个状态值（这里是hasRunThree），每次流发出新值时都会基于上一次的状态计算新状态。
• 初始状态设为false，表示还没执行过this.three()。
• 每次queryParams更新时，先执行this.one()和this.two()；如果是第一次触发，额外执行this.three()并将状态改为true，后续触发就不会再执行this.three()了。
• takeUntil(this.destroy$)确保组件销毁时自动取消订阅，避免内存泄漏。

方案二：使用share共享源流 + merge合并逻辑 #

这种方案先共享源流（避免重复订阅queryParams），再拆分逻辑后合并成一个订阅：

// 共享源流，确保只订阅一次queryParams
const queryParams$ = this.route.queryParams.pipe(
  takeUntil(this.destroy$),
  share()
);

// 合并两个逻辑流，最终只需要一个订阅
merge(
  // 处理每次都要执行的one和two
  queryParams$.pipe(tap(() => {
    this.one();
    this.two();
  })),
  // 处理仅执行一次的three
  queryParams$.pipe(first(), tap(() => this.three()))
).subscribe();

逻辑说明： #

• share()操作符让多个订阅共享同一个源流的订阅，避免重复触发route.queryParams的内部逻辑。
• merge操作符将两个子流合并成一个：第一个子流每次queryParams变化都触发one和two；第二个子流通过first()只取第一次的queryParams并触发three。
• 最终只需要一个subscribe()调用，同时保留了原有的所有执行逻辑。

这两种方案都能满足你的需求，你可以根据代码风格偏好选择其中一种～

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