RxJS中单段.pipe与多段嵌套.pipe的执行逻辑差异咨询

首先你遇到的问题和「单段/多段pipe」本身没有关系，同一层级下obs.pipe(op1).pipe(op2)和obs.pipe(op1, op2)的执行逻辑完全等价，本质差异是你调整了两个repeatWhen操作符的嵌套层级，解决了平级串联时的complete事件拦截问题。

首先明确repeatWhen的核心工作逻辑 #

repeatWhen只会在**上游Observable完成（complete）**时才会激活你传入的notifier触发器，只要notifier发射值，就会重新订阅上游；如果上游永远不触发complete，repeatWhen的触发器永远不会生效。

第一种平级写法失效的原因 #

我们拆解下你第一种写法的流结构：

public readonly initialValueFlat$ = this.seed$
  .pipe(
    switchMap(seed => defer(
      () => {
        this._isLoading$$.next(true);
        return this.getFakeData(seed)
          .pipe(
            finalize(() => this._isLoading$$.next(false)));
      })), // 上游1：请求完成后会触发complete
    repeatWhen(x => this._reloadTrigger$), // 操作符1：监听上游1的complete
    switchMap(x => of(x)), // 上游2：of(x)完成后会触发complete
    repeatWhen(x => this._resetTrigger$), // 操作符2：监听上游2的complete
    shareReplay(1));

这里的致命问题是：repeatWhen收到上游的complete事件后，不会把complete向下游传递，而是直接重新订阅上游。

• 操作符1（重载触发器对应的repeatWhen）会拦截上游请求流的complete事件，只会在收到重载信号时重新发起请求，永远不会向下游传递complete
• 操作符2（重置触发器对应的repeatWhen）的上游永远收不到complete事件，所以它的notifier永远不会被激活，重置触发器完全不生效

第二种嵌套写法生效的原因 #

你调整后的写法把两个repeatWhen放到了不同的嵌套作用域，各自的上游都能正常触发complete，不会被拦截：

public readonly initialValue$ = this.seed$
  .pipe(
    switchMap(seed => defer(
      () => {
        this._isLoading$$.next(true);
        return this.getFakeData(seed)
          .pipe(
            finalize(() => this._isLoading$$.next(false)));
      }) // 子上游1：请求完成后触发complete
      .pipe(
        repeatWhen(x => this._reloadTrigger$) // 子操作符1：只监听子上游1的complete，拦截的complete不会流出switchMap
      )
      .pipe(
        switchMap(x =>
          of(x) // 子上游2：发射x后立即触发complete
            .pipe(
              repeatWhen(x => this._resetTrigger$)) // 子操作符2：只监听子上游2的complete，能正常响应重置信号
        ))),
    shareReplay(1));

• 重载用的repeatWhen只作用于内部的请求流，拦截的complete不会影响外层其他逻辑，每次重载触发就重新发起请求，返回新值
• 重置用的repeatWhen只作用于of(x)这个小流，of(x)会立即complete所以触发器能正常激活，每次重置触发就重新发射当前最新的x，不会发起新请求
• 外层的shareReplay(1)会缓存最新值，晚订阅的观察者可以直接拿到缓存值，同时能正常响应后续的重载、重置事件

可选优化写法 #

可以把逻辑拆得更清晰，避免过深的嵌套：

public readonly initialValue$ = this.seed$.pipe(
  switchMap(seed => 
    // 处理重载逻辑
    this._reloadTrigger$.pipe(
      startWith(void 0), // 初始自动触发一次请求
      switchMap(() => {
        this._isLoading$$.next(true);
        return this.getFakeData(seed).pipe(finalize(() => this._isLoading$$.next(false)));
      })
    )
  ),
  // 处理重置逻辑：每次触发重置就重新发射最新值
  switchMap(val => this._resetTrigger$.pipe(startWith(val), mapTo(val))),
  shareReplay(1)
);

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