Angular7 + RxJs 多个订阅者到同一个 Observable

Question

这可能是一个菜鸟问题，但我可以找到它的参考。

我有两个组件放置在不同的地方，每个组件都使用一个服务来查询端点并获取一些数据（例如用户配置文件）。

服务返回一个 observable，我需要 observable 是唯一的，或者至少要发出一个唯一的请求。

// Service, shared across multiple components
@Injectable()
export class SomeService {
    getData(): Observable<DataModel> {
        return this._http.get<DataResponse>('/some-route').pipe(
          map((response: DataResponse) => this.handleResponse(response)),
          share() // is this right ?
        );
    }

}

// Component requesting data
@Component({ selector: '...', ... })
export class FirstComponent implements OnInit {
  constructor(private _service: SomeService) { }

    ngOnInit() {
        this._service.getData().subscribe(
          data => {
            console.log(data);
          }
        );
      }
}

// Another component requesting the same data
@Component({ selector: '...', ... })
export class SecondComponent implements OnInit {
  constructor(private _service: SomeService) { }

    ngOnInit() {
        this._service.getData().subscribe(
          data => {
            console.log(data);
          }
        );
      }
}

该服务工作并获取数据，但请求发送了两次，我只希望发送一个请求。这些组件同时存在（假设一个在屏幕顶部，第二个在屏幕底部）。所以他们同时提出请求。

有没有办法让服务只发送一个请求。

顺便说一句，第一个请求的状态为 200，第二个请求的状态为 304。

谢谢。

更新

可能的解决方案

到目前为止，我通过添加一个服务变量来管理

private _observable: Observable<DataModel>;

那么当获取数据时

getData(): Observable<DataModel> {
    if (!this._observable) {
        this._observable = this._http.get<DataResponse>('/some-route').pipe(
          map((response: DataResponse) => this.handleResponse(response)),
          // HERE IS THE TRICK
          publishLast(),
          refCount()
        )
    }

    return this_observable;
}

诀窍是使用 publishLast 和 refCount

有更好的方法/想法吗？

Answer 1

您可以轻松地将您的请求分组到一个 Observable 中，该 Observable 将在您使用 forkJoin 运算符完成所有请求时解决

这是一个例子

  public requestDataFromMultipleSources(): Observable<any[]> {
    let response1 = this.http.get(requestUrl1);
    let response2 = this.http.get(requestUrl2);
    let response3 = this.http.get(requestUrl3);
    // Observable.forkJoin (RxJS 5) changes to just forkJoin() in RxJS 6
    return forkJoin([response1, response2, response3]);
  }

在您的示例中，您应该在父组件中提出请求。然后将结果注入到子组件中

Answer 2

看看我的库 ngx-rxcache。它简化了状态管理。

https://github.com/adriandavidbrand/ngx-rxcache

@Injectable()
export class SomeService {
 private dataCache = this.cache.get<DataModel>({
    id: 'some uniquie id',
    load: true,
    construct: () => this.http.get<DataResponse>('/some-route').pipe(
      map((response: DataResponse) => this.handleResponse(response))
    )
 });

 data$ = this.dataCache.value$;

 constructor(cache: RxCacheService, http: HttpClient) {}
}

然后你就可以访问服务上的数据了

this.service.data$;

https://medium.com/@adrianbrand/angular-state-management-with-rxcache-468a865fc3fb

Answer 3

您可以改用 shareReplay，它将从可观察源重播事件给所有订阅者，并确保您只分享最后一个结果

 shareReplay(1)

所以你的代码看起来像这样

getData(): Observable<DataModel> {
    return this._http.get<DataResponse>('/some-route').pipe(
      map((response: DataResponse) => this.handleResponse(response)),
      shareReplay(1) 
    );
}

Answer 4

我发现以下模式对此非常有效：

@Injectable()
export class SomeService {
  myDataObservable$: Observable<DataModel>;

  constructor( ) {
    this.myDataObservable$ = this.getData().pipe(shareReplay());
  }

  getData(): Observable<DataModel> {
    return this._http.get<DataResponse>('/some-route').pipe(
      map((response: DataResponse) => this.handleResponse(response))
    );
  }
}

1. 在服务上创建一个属性 (myDataObservable$)，该属性包含多个组件将使用的 observable。
2. 拥有一个执行 http 请求的方法 (getData())。
3. 在构造函数中，调用方法并添加shareReplay()。
4. 在组件中，您将其与“this.someService.myDataObservable$.subscribe(...)”一起使用

使用这种模式，在您至少在一个组件中调用 myDataObservable$ 上的 subscribe 之前，不会发送 http 请求。在第一个组件调用 subscribe 后，所有后续订阅都将使用已经存在的值，因此您不会有多个 http 请求。