Angular中如何在单个RxJS流中刷新并监听刷新后的数据？

问题背景 #

核心代码如下：

data$:Observable = this.dataRefresh$.pipe(switchMap(() => 
  this.http.get('url')), shareReplay());
dataRefresh$:Subject = new Subject();

ngOnInit(){
  this.externalService.getExternalTrigger()
    .pipe(
      // TODO - trigger data refresh and switch to data$
      tap(data => this.doThings(data)))
    .subscribe();
}

refresh(){
  this.refreshData$.next(); // 注意此处变量名拼写错误
}

需求是外部服务通知服务器数据变更后获取新数据，因多组件复用数据使用了shareReplay，但尝试构建如下流时出现循环依赖错误：

combineLatest([
   of().pipe(tap(() => this.refresh)),
   this.data.pipe(skipUntil(this.refreshData$))
 ]).pipe(map(([,data]) => data));

解决步骤 #

1. 修正变量名拼写错误 #

首先修正refresh()方法里的变量名错误，避免逻辑异常：

refresh(){
  this.dataRefresh$.next();
}

2. 简化外部触发逻辑，消除循环根源 #

无需使用combineLatest构建复杂流，直接在外部触发的流中完成「触发刷新+获取新数据」的逻辑：

ngOnInit(){
  this.externalService.getExternalTrigger()
    .pipe(
      tap(data => this.doThings(data)),
      // 触发刷新后切换到最新的data$
      switchMap(() => {
        this.dataRefresh$.next();
        return this.data$;
      })
    )
    .subscribe(newData => {
      // 处理刷新后的新数据
    });
}

这个逻辑直接明了：外部触发事件→执行前置操作→触发数据刷新→获取并订阅最新数据，完全不存在循环依赖。

3. 优化data$的初始化（可选） #

如果希望页面加载时自动获取初始数据，将Subject改为BehaviorSubject并赋予初始值，这样data$在首次订阅时会自动触发HTTP请求：

dataRefresh$:BehaviorSubject<void> = new BehaviorSubject<void>(undefined);

此时无需在组件初始化时手动调用refresh()。

4. 移除不必要的skipUntil操作 #

data$本身由dataRefresh$触发，每次dataRefresh$.next()都会让data$发出最新请求结果，直接订阅data$即可获取最新数据，之前的skipUntil(this.refreshData$)属于冗余操作，只会增加逻辑复杂度。

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