如何正确管理RxJS observable中的内部状态

你原有方案出现异常的核心原因有两个：

• 把可变状态globalDeletedIds定义在流的外部，只要返回的Observable被多次订阅，所有订阅都会往同一个全局数组写入数据，造成状态交叉污染
• combineLatest的执行时序要求所有上游流都发出至少一个值后才会触发回调，若初始Promise resolve前就收到删除变更，这部分变更会直接丢失，不会被记录

RxJS中做流内持久化状态管理的标准操作符是scan，它会在流的生命周期内独立维护内部状态，每次订阅都会生成全新的状态副本，完全不会暴露到流外部造成污染，也不会漏掉任何上游推送的事件。

基础实现（逻辑直观易扩展） #

直接把当前有效项目列表作为维护的状态，所有变更逻辑都收敛在状态更新环节：

import { from } from "rxjs";
import { scan, switchMap, map } from "rxjs/operators";

const initialItems$ = from(initialPromise);

return initialItems$.pipe(
  // 初始列表加载完成后，再接入变更流
  switchMap(initialItems => mutationObservable.pipe(
    // scan第一个参数是状态更新函数，第二个参数是初始状态
    scan((currentItems, mutation) => {
      // 处理删除变更，其他类型变更（新增/编辑）可在此扩展
      if (mutation?.deleted) {
        // 遵循不可变原则，返回新数组，不修改原状态引用
        return currentItems.filter(item => item.id !== mutation.id);
      }
      // 不匹配的变更类型直接返回原状态
      return currentItems;
    }, initialItems),
    // 基于最新状态生成最终结果
    map(currentItems => buildResult(currentItems))
  ))
);

大数据量优化实现 #

如果项目列表量级很大，每次全量过滤性能较差，可以单独维护已删除ID的集合作为状态，利用Set的O(1)查询效率提升性能：

import { from } from "rxjs";
import { scan, switchMap, map } from "rxjs/operators";

const initialItems$ = from(initialPromise);

return initialItems$.pipe(
  switchMap(initialItems => mutationObservable.pipe(
    scan((deletedSet, mutation) => {
      if (mutation?.deleted) {
        return new Set([...deletedSet, mutation.id]);
      }
      return deletedSet;
    }, new Set<string>()),
    map(deletedSet => {
      const currentItems = initialItems.filter(item => !deletedSet.has(item.id));
      return buildResult(currentItems);
    })
  ))
);

方案优势 #

• 无任何外部可变状态，多次订阅不会出现状态污染
• 不存在时序漏洞，初始数据加载前收到的所有变更都会被scan按顺序接收处理，不会丢失
• 所有状态更新逻辑完全收敛，后续新增编辑、新增类的变更需求时，直接在scan的回调函数中扩展对应逻辑即可，维护成本低

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