Rxjs Observable 没有向所有订阅者发送最新数据答案

【问题标题】：Rxjs Observable not emitting latest data to all the subscribersRxjs Observable 没有向所有订阅者发送最新数据
【发布时间】：2020-06-11 03:54:52
【问题描述】：

我已经声明了一个服务中的主题，并尝试通过使用“subject.next()”对后端进行 GET 调用来推送响应。还使用“subject.asObservable()”为该主题分配了一个可观察对象并订阅了两个不同组件（父级和路由器出口子级）中的 observable。问题是 observable 只为一个订阅者而不是为两个订阅者发出新值。

服务

subject = new Subject<any>();
  observable = this.subject.asObservable();


 public getDetails(id) {
    return this.http.get(this.url.concat(id))
      .pipe(map(data => this.extractDetails(data),
            catchError(err => this.handleError(err)));
  }
 public extractDetails(data){
    this.subject.next(data);
    return data;
  }

父组件

ngOnInIt(){
 this.hitUrl();

}
getDetails(){
    this.service.observable.subscribe(
      (data: any) => {
        console.log(data);
      },
      err => {
        console.error(err);
      }
    )
  }

  hitUrl(id: string) {
    this.service.getDetails(id).subscribe(
      (data: any) => 
        if(!!data){
            console.log(data);
            this.getDetails();
          }
      },
      err => {
        console.log(err);
      }
    );
  }

路由器插座子组件

ngOnInIt(){
 this.hitUrl();

}
getDetails(){
    this.service.observable.subscribe(
      (data: any) => {
        console.log(data);
      },
      err => {
        console.error(err);
      }
    )
  }

  hitUrl(id: string) {
    this.service.getDetails(id).subscribe(
      (data: any) => 
        if(!!data){
            console.log(data);
            this.getDetails();
          }
      },
      err => {
        console.log(err);
      }
    );
  }

在组件中，hitUrl() 也会在执行号召性用语时触发，然后我需要 observable 向两个订阅者发出最新数据（如果从父级或路由器出口子级执行号召性用语）。

【问题讨论】：

标签： angular typescript rxjs observable

【解决方案1】：

你混淆了概念。在您的 service.getDetails 函数中，每次调用该函数时都会返回一个新的 observable。因此，您的 observable 属性未使用。您应该只导出observable 属性并保持getDetails 私有，调用该函数只是为了获取数据。

如何创建服务缓存

行为主题

您可以使用 BehaviorSubject，它是 Subject 的一种变体，它需要一个初始值并在订阅时发出其当前值。

private subject = new BehaviorSubject<any>();
private observable = this.subject.asObservable();

  public getDetails(id) {
    if (!this.observable) {
      this.loadData();
    }
    return this.observable;
  }

  private loadData() {
    this.http.get(this.url.concat(id))
    .subscribe(data => {
      this.subject.next(data);
    })
    this.observable = this.subject.asObservable();
  }

shareReplay

最好的方法是使用 rxjs shareReplay。这个操作符返回一个 Observable，它共享一个对底层源的订阅。换句话说，让我们的 observable 变热。

const CACHE_SIZE = 1;
class Service { 

  private _data$: Observable<YourType>;

  get data(): Observable<YourType> {
    if (!this._data$) {
      this._data$ = anyHttpCall()
        .pipe(shareReplay({ refCount: true, bufferSize: CACHE_SIZE })
      );
    }
    return this._data$;
  }
}

bufferSize 决定了重放缓冲区的最大元素数，即为每个订阅者缓存和重放的元素数。

这篇文章很好地解释了这一点：https://blog.thoughtram.io/angular/2018/03/05/advanced-caching-with-rxjs.html

【讨论】：