在RxJS中如何用声明式方式避免订阅与BehaviorSubject.value

问题背景 #

我在重构Angular应用，目标是移除所有手动订阅，完全依赖Angular的async pipe（用声明式代替命令式）。但遇到多数据源触发状态变化的场景时，很难写出简洁的声明式代码——单数据源的话直接用scan就能搞定，但多操作（增删）加初始数据的组合让我犯难。

场景需求 #

做一个简单组件：

• 路由解析阶段获取字符串数组作为初始数据
• 组件展示列表，支持通过按钮增删列表项

限制条件 #

• 禁止手动调用subscribe，完全靠async pipe自动管理订阅/取消订阅
• 禁止使用BehaviorSubject.value，避免命令式操作
• 仅允许用Subject传递按钮点击事件，不能额外引入其他Subject，所有状态通过现有Observable组合实现

现有尝试方案（均有弊端） #

• BehaviorSubject + .value：直接用Subject存状态，操作时取.value生成新数组——但这是命令式写法，不符合要求
• Redux风格Action + scan：定义Action接口，按钮点击发射对应Action，在scan里判断Action类型处理逻辑——但需要混合不同值类型，代码冗余
• shareReplay + concatMap：用shareReplay(1)模拟BehaviorSubject，通过concatMap切换到当前状态Observable并take(1)避免循环，实现了需求但代码太复杂，具体实现如下：

现有代码 #

list-view.component.html

<ul>
  <li *ngFor="let item of items$ | async; let i = index">
    {{ item }} <button (click)="remove$.next(i)">remove</button>
  </li>
</ul>

<button (click)="add$.next('test2')">add</button>

list-view.component.ts

// 传递添加按钮点击事件的Subject，参数为新增的字符串
add$ = new Subject<string>();

// 传递删除按钮点击事件的Subject，参数为要删除项的索引
remove$ = new Subject<number>();

// 用于展示的列表Observable
items$: Observable<string[]>;

constructor(private readonly _route: ActivatedRoute) {
  this.items$ = merge(
    // 初始数据：路由解析的数组
    this._route.data.pipe(map((items) => items[ITEMS_KEY])),

    // 添加操作：基于当前列表生成新数组
    this.add$.pipe(
      concatMap((added) =>
        this.items$.pipe(
          map((list) => [...list, added]),
          take(1)
        )
      )
    ),

    // 删除操作：基于当前列表生成新数组
    this.remove$.pipe(
      concatMap((index) =>
        this.items$.pipe(
          map((list) => [...list.slice(0, index), ...list.slice(index + 1)]),
          take(1)
        )
      )
    )
  ).pipe(shareReplay(1));
}

更简洁的声明式解决方案 #

核心思路是把所有状态变更操作统一转换成对当前状态的更新函数，用scan累积状态，避免循环引用和复杂的切换操作：

优化后的list-view.component.ts #

add$ = new Subject<string>();
remove$ = new Subject<number>();
items$: Observable<string[]>;

constructor(private readonly _route: ActivatedRoute) {
  // 把所有操作转换成更新函数：接收当前状态，返回新状态
  const updateActions$ = merge(
    // 初始数据：直接返回初始数组作为第一个状态
    this._route.data.pipe(
      map((data) => (current: string[]) => data[ITEMS_KEY])
    ),
    // 添加操作：返回添加新项的更新函数
    this.add$.pipe(
      map((newItem) => (current: string[]) => [...current, newItem])
    ),
    // 删除操作：返回删除指定索引项的更新函数
    this.remove$.pipe(
      map((index) => (current: string[]) => [
        ...current.slice(0, index),
        ...current.slice(index + 1)
      ])
    )
  );

  // 用scan累积状态，初始值设为空数组，初始数据会覆盖它
  this.items$ = updateActions$.pipe(
    scan((currentState, updateFn) => updateFn(currentState), [])
  );
}

优势说明 #

• 完全声明式：所有状态变更都是纯函数，没有命令式的状态读取/修改
• 代码简洁：统一用updateFn处理所有操作，scan负责累积状态，逻辑清晰
• 无循环引用：不需要引用items$自身来生成新状态，避免了原方案中concatMap+take(1)的复杂处理
• 符合所有限制：没有手动订阅，没有用BehaviorSubject.value，仅用了必要的Subject传递点击事件

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