如何实现ForkJoin获取NgRx Store数据时满足指定值后执行方法？

方案1：利用combineLatest监听状态变化（推荐，符合NgRx响应式思路） #

这种方案无需手动重试，通过持续监听Store状态变化，直到storeSelector2的值变为'50'时自动触发目标方法，更贴合RxJS与NgRx的响应式设计，仅在Store状态更新时触发检查，性能更优。

代码示例：

import { combineLatest, take, filter } from 'rxjs';

public ngOnInit(): void {
  combineLatest([
    this.myNgrxStore.select(storeSelector1),
    this.myNgrxStore.select(storeSelector2),
    this.myNgrxStore.select(storeSelector3),
  ]).pipe(
    // 筛选出storeSelector2等于'50'的状态组合
    filter(([data1, data2, data3]) => data2 === '50'),
    // 获取到符合条件的数据后立即取消订阅
    take(1)
  ).subscribe(([data1, data2, data3]) => {
    this.randomMethod(data1, data2, data3);
  });
}

说明： #

• combineLatest会在任意一个Selector的状态更新时，发出三个数据的最新组合。
• filter操作符确保只有第二个数据符合要求时，才会继续向下传递数据流。
• take(1)保证获取到目标数据后自动终止订阅，避免不必要的资源占用。

方案2：手动轮询重试（适合固定间隔检查的场景） #

如果业务需要每隔固定时间主动检查Store状态，而非被动监听变化，可以用timer结合forkJoin实现轮询逻辑：

代码示例：

import { forkJoin, timer, take, filter, switchMap } from 'rxjs';

public ngOnInit(): void {
  timer(0, 1000) // 立即执行第一次检查，之后每1000ms执行一次
    .pipe(
      // 每次定时器触发时，重新获取三个Selector的当前值
      switchMap(() => forkJoin([
        this.myNgrxStore.select(storeSelector1).pipe(take(1)),
        this.myNgrxStore.select(storeSelector2).pipe(take(1)),
        this.myNgrxStore.select(storeSelector3).pipe(take(1)),
      ])),
      // 筛选符合条件的数据组合
      filter(([data1, data2, data3]) => data2 === '50'),
      // 获取到目标数据后停止轮询
      take(1)
    ).subscribe(([data1, data2, data3]) => {
      this.randomMethod(data1, data2, data3);
    });
}

说明： #

• timer(0, 1000)指定首次检查立即执行，后续每隔1秒触发一次。
• switchMap会在每次定时器触发时，重新执行forkJoin获取Store的最新数据。
• 一旦筛选到符合条件的数据，take(1)会终止整个数据流，停止轮询。

额外注意事项 #

• 组件销毁时需取消订阅，避免内存泄漏，可通过Subscription对象管理：

  private subscription!: Subscription;
  
  ngOnInit(): void {
    this.subscription = combineLatest(/* ... */).subscribe(/* ... */);
  }
  
  ngOnDestroy(): void {
    this.subscription.unsubscribe();
  }
  

• 若存在storeSelector2永远无法变为'50'的情况，建议添加超时逻辑：

  combineLatest(/* ... */).pipe(
    filter(/* ... */),
    take(1),
    timeout(30000) // 30秒未获取到符合条件的数据则抛出超时错误
  ).subscribe({
    next: (data) => this.randomMethod(...data),
    error: (err) => console.error('等待超时：', err)
  });
  

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