如何手动更新组合Observable中的notifications$部分？

我有一个绑定到async pipe的组合Observable combinedNotifications$，首次加载正常，但希望能仅手动更新notifications$。之前对Observable了解不多，之前的用例都更简单。

combinedNotifications$: Observable<any> = combineLatest([
    this.notifications$,
    this.notificationTypes$
  ]).pipe(
    map((i: any) => ({
      tableData: i[0],
      notificationTypes: i[1]
    })),
    map((f: any) => {
      return ({
        tableData: {
          ...f.tableData, content: f.tableData.content.map((v: any) => ({
            ...v,
            type: f.notificationTypes.eventTypes.find((x: any) => x.value === v.type)?.label,
            cause: f.notificationTypes.causeTypes.find((x: any) => x.value === v.cause)?.label
          }))
        },
        notificationTypes: f.notificationTypes
      })
    }),
    tap((d) => console.log(d)),
    shareReplay(1)
  );

我有一个带分页交互的表格，在创建这个组合Observable之前，我可以单独更新通知：

retrieveNotificationsPage(queryArgs: Pageable = {}, searchData: any = {}) {
    this._loading$.next(true);
    const actionAuditLogsPage$ = this.apiService
      .retrieveNotificationsView$(queryArgs, { ...searchData })
      .pipe(
        delay(300),
        tap({
          complete: () => {
            this._loading$.next(false);
          },
          error: (err) => {
            this._loading$.next(false);
            throw err;
          }
        }),
        catchError((err) => of({ error: err })),
        // shareReplay()
      );
    return actionAuditLogsPage$;
  }

调用方式如下：

loadRecords(event: LazyLoadEvent) {
    ...
    this.notifications$ = this.retrieveNotificationsPage({
      page: isNaN(selectedPage) ? 0 : selectedPage,
      ...(maxPageRowSize != null ? { size: maxPageRowSize } : {})
    }, this.searchForm.value);
  }

我试过用merge实现，但陷入了无限循环：

this.combinedNotifications$ = merge(notificationPage$, this.combinedNotifications$);

解决方法 #

核心问题是直接重新赋值notifications$会破坏原有的combineLatest订阅链，而merge的方式又导致了循环订阅。正确的做法是把notifications$改成BehaviorSubject（或ReplaySubject），通过next方法手动推送新数据，而非重新赋值Observable。

步骤如下：

1. 将notifications$从普通Observable改为BehaviorSubject：

// 初始化时可传入默认值，比如空数据结构
private notificationsSubject = new BehaviorSubject<any>({ content: [] });
notifications$ = this.notificationsSubject.asObservable();

2. 修改loadRecords方法，不再重新赋值notifications$，而是调用next推送新的分页数据：

loadRecords(event: LazyLoadEvent) {
    ...
    this.retrieveNotificationsPage({
      page: isNaN(selectedPage) ? 0 : selectedPage,
      ...(maxPageRowSize != null ? { size: maxPageRowSize } : {})
    }, this.searchForm.value).subscribe(data => {
      this.notificationsSubject.next(data);
    });
}

3. 保留原有的combinedNotifications$定义，无需修改：
   它会自动监听notifications$的变化，当你通过notificationsSubject.next推送新数据时，combineLatest会重新执行映射逻辑，更新视图。

4. （可选）组件销毁时清理资源：

ngOnDestroy() {
  this.notificationsSubject.complete();
}

这样就能实现手动触发notifications$更新，同时避免无限循环，原有的async pipe绑定也能正常工作。

内容的提问来源于stack exchange，提问作者ak.leimrey