angular 等待几个订阅做某事

Question

在一个组件中，我订阅了两个服务。第一个，

this.myFirstService.currentData.subscribe()

其中 currentData 来自 BehaviorSubject

  private currentDataSource = new BehaviorSubject<any>([]);
  currentData = this.currentDataSource.asObservable();

第二个来自 http.get

  getData(): Observable<Data> {
  if(localStorage.getItem('token'))
    var headers = new HttpHeaders({'Authorization':'Bearer ' +   localStorage.getItem('token')});
  else
    var headers = new HttpHeaders({'Authorization':'Anonymous '});

  return this.http.get<Data>("http://someurl/",{
    headers:headers,
    responseType: 'json'
  })
    .pipe();
}

我想合并从两个订阅接收的数据。我该如何存档？

Answer 1

由于您使用的是 RxJs，因此有几种方法可以做到这一点:

康卡特

Concat 会将两个 observable 组合成一个组合序列，但第二个 observable 在第一个完成之前不会开始发射。

let first = Observable.timer(10,500).map(r => {
  return {source:1,value:r};
}).take(4);
let second = Observable.timer(10,500).map(r => {
  return {source:2,value:r};
}).take(4);
first.concat(second).subscribe(res => this.concatStream.push(res));

这将合并两者，但您将在第二个之前收到第一个可观察结果:

0 1 2 3 0 1 2 3

合并

Merge 与 concat 类似，但它会交错发出的值，而不是在开始第二个 observable 之前完成第一个 observable。

let first = Observable.timer(10,500).map(r => {
  return {source:1,value:r};
}).take(4);
let second = Observable.timer(10,500).map(r => {
  return {source:2,value:r};
}).take(4);
first.merge(second).subscribe(res => this.mergeStream.push(res));

你会得到:

0 0 1 1 2 2 3 3 

叉连接

我们使用 forkJoin 并行执行 observable。一个常见的用例是并行发出多个 http 请求。在我的示例中，我 forkJoining 两个非常简单的 observables，但关键是订阅者在两个 observables 完成之前不会收到任何值。

let first = Observable.of({source:1,value:1});
let second = Observable.of({source:2,value:1});
Observable.forkJoin(first,second)
            .subscribe((res:Array) => this.forkJoinStream = res);

这类似于 Angular 1.x 中的 $q.all()。

最后

平面图

如果您的 observables 之间存在依赖关系，请使用此选项。假设您必须先获取用户，然后才能通过两个不同的 http 请求获取用户详细信息:

let first = Observable.of(10);
first.flatMap((operand1) => {
  return Observable.of(operand1 + 10);
})
.subscribe(res => this.flatMappedStreams = {msg: '10 + 10 = ' + res});

你会得到:

10 + 10 = 20