RxJS combineLatest如何仅对单个observable添加timer轮询

问题背景 #

现有两个业务实体：

• Games（游戏）接口结构：

{
  id: string,
  name: string
}

• Jackpots（奖池）接口结构：

{
  game: string,
  amount: number
}

原有实现通过timer做轮询时，会同时触发游戏列表和奖池列表两个接口的重复请求，需求是仅定时轮询奖池接口，游戏列表仅在初始化时拉取一次，最终按id关联将奖池数据挂载到对应游戏条目下。
原问题代码：

private getGames = (category: string): Observable<IGame[]> => {
        return timer(0, 3000).pipe(
            mergeMap(() => combineLatest([this.gameService.getGames(), this.gameService.getJackpots()])),
            map(([games, jackpots]) => {
                return games?.filter(game => game?.categories?.includes(category))?.map(game => {
                    return { ...game, jackpot: jackpots?.find(jackpot => jackpot?.game === game?.id) };
                })
            })
        );
    };

问题原因 #

原代码将两个接口的请求逻辑全部放在timer触发的回调函数内，每次定时器发射值都会重新执行两个接口的请求，因此游戏列表会被重复拉取。

修复代码 #

核心思路是拆分两个数据流：游戏列表只在订阅时请求一次，奖池数据走定时轮询，最后通过combineLatest实现数据联动更新。

private getGames = (category: string): Observable<IGame[]> => {
  // 游戏列表仅拉取一次，提前完成分类过滤，避免重复计算
  const filteredGames$ = this.gameService.getGames().pipe(
    map(games => games?.filter(game => game?.categories?.includes(category)) ?? [])
  );

  // 奖池数据按配置定时轮询：首次立即执行，之后每3秒拉取一次
  const jackpots$ = timer(0, 3000).pipe(
    mergeMap(() => this.gameService.getJackpots())
  );

  // 组合两个流：游戏列表返回后，每次奖池更新都会自动触发数据合并
  return combineLatest([filteredGames$, jackpots$]).pipe(
    map(([games, jackpots]) => {
      return games.map(game => ({
        ...game,
        jackpot: jackpots?.find(jackpot => jackpot.game === game.id)
      }))
    })
  );
};

逻辑说明 #

• 游戏列表接口仅在订阅当前方法返回的Observable时触发1次，不会跟随轮询重复请求
• 奖池接口完全按照定时器规则执行，首次加载立即拉取，后续每3秒刷新一次数据
• 每次奖池数据返回后，都会自动和已拉取的游戏列表做匹配，更新每个游戏的奖池字段
• 分类过滤逻辑提前到游戏流中处理，每次轮询奖池时不需要重复遍历全量游戏列表做过滤，减少无意义的性能消耗
  如果后续需要支持游戏列表的刷新（比如切换分类、手动触发刷新），只需要替换filteredGames$的数据源为对应触发流即可，奖池轮询逻辑不需要改动。

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