如何处理NgRx Component Store的多Observable并发更新问题

问题场景 #

当多个Observable同时触发更新，且都作用于NgRx Component Store的同一属性时，会出现状态覆盖问题。以电影显示场景为例：

组件中每个电影的可见性变更由独立Observable触发，代码如下：

someMovie.pipe(takeUntil(this.onDestroy$)).subscribe(({movie: IMovie, shouldBeDisplayed: boolean}) => {
  this.updateMovieVisibility(movie, shouldBeDisplayed);
});

Component Store的更新逻辑：

public updateMovieVisibility(movie: IMovie, shouldBeDisplayed: boolean) {
  this.patchState((state) => {
    let newMoviesToDisplay = [...state.displayedMovies];

    if (shouldBeDisplayed) {
      newMoviesToDisplay.push(movie);
    } else {
      newMoviesToDisplay = newMoviesToDisplay.filter((item) => item.id !== movie.id);
    }

    return {
      displayedMovies: newMoviesToDisplay,
    };
  });
}

问题表现 #

当电影Y和Z的Observable几乎同时发射值时：

1. 初始状态：displayedMovies = [X]
2. 电影Y的回调执行，基于旧状态生成[X,Y]并更新
3. 电影Z的回调在Y的状态更新生效前就读取了旧状态，生成[X,Z]并覆盖更新
4. 最终状态为[X,Z]，丢失了电影Y的显示状态

解决方案 #

方案1：合并Observable，统一计算最终状态 #

将所有电影的可见性Observable合并，每次任何一个电影的状态变化时，统一计算所有需要显示的电影，避免并发更新冲突。

代码示例：

// 收集所有电影的可见性Observable
const movieVisibilityObservables = [movieYVisibility$, movieZVisibility$, ...其他电影Observable];

// 合并Observable，每次变化时计算最终显示列表
combineLatest(movieVisibilityObservables).pipe(
  takeUntil(this.onDestroy$)
).subscribe((visibilityStates) => {
  // 筛选出所有需要显示的电影
  const displayedMovies = visibilityStates
    .filter(({ shouldBeDisplayed }) => shouldBeDisplayed)
    .map(({ movie }) => movie);
  
  this.patchState({ displayedMovies });
});

方案2：串行化更新请求 #

通过Subject收集所有更新请求，使用concatMap确保前一个状态更新完成后，再处理下一个请求，保证每次更新都基于最新状态。

代码示例：

// 创建Subject接收更新请求
private updateRequests$ = new Subject<{movie: IMovie, shouldBeDisplayed: boolean}>();

constructor() {
  // 串行处理所有更新请求
  this.updateRequests$.pipe(
    concatMap(req => {
      this.updateMovieVisibility(req.movie, req.shouldBeDisplayed);
      return of(null);
    }),
    takeUntil(this.onDestroy$)
  ).subscribe();
}

// 原订阅逻辑改为发送请求到Subject
someMovie.pipe(takeUntil(this.onDestroy$)).subscribe(req => {
  this.updateRequests$.next(req);
});

// 原更新函数保持不变
public updateMovieVisibility(movie: IMovie, shouldBeDisplayed: boolean) {
  this.patchState((state) => {
    let newMoviesToDisplay = [...state.displayedMovies];

    if (shouldBeDisplayed) {
      newMoviesToDisplay.push(movie);
    } else {
      newMoviesToDisplay = newMoviesToDisplay.filter((item) => item.id !== movie.id);
    }

    return {
      displayedMovies: newMoviesToDisplay,
    };
  });
}

方案3：利用Component Store Effect的串行性 #

NgRx Component Store的Effect默认串行执行，可将更新逻辑绑定到Effect中，确保请求按顺序处理。

代码示例：

// 定义更新器
readonly updateMovieVisibility = this.updater((state, action: {movie: IMovie, shouldBeDisplayed: boolean}) => {
  let newMoviesToDisplay = [...state.displayedMovies];

  if (action.shouldBeDisplayed) {
    newMoviesToDisplay.push(action.movie);
  } else {
    newMoviesToDisplay = newMoviesToDisplay.filter((item) => item.id !== action.movie.id);
  }

  return {
    displayedMovies: newMoviesToDisplay,
  };
});

// 创建Effect串行处理更新请求
private updateEffect$ = this.effect((actions$) => {
  return actions$.pipe(
    concatMap(action => {
      this.updateMovieVisibility(action);
      return EMPTY;
    })
  );
});

// 原订阅逻辑改为触发Effect
someMovie.pipe(takeUntil(this.onDestroy$)).subscribe(req => {
  this.updateEffect$.next(req);
});

问题根源 #

JavaScript是单线程执行环境，当多个Observable几乎同时发射值时，它们的回调会在同一事件循环中执行，都读取了同一个旧的状态快照。各自修改后，后执行的patchState会覆盖先执行的结果，导致状态丢失。

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