RxJS Service如何同时执行请求并逐个返回响应而非等待全部完成？

你现在遇到的问题正好是forkJoin的特性带来的——它必须等所有Observable都完成才会一次性返回所有结果。而你需要的是每个请求完成就立即处理响应，这时候用merge操作符就完美适配你的需求，它和forkJoin一样会并行发起所有请求，但只要有一个请求完成，就会立刻把该结果推送给下游订阅者。

具体改造步骤 #

假设你原来的代码大概是这样的：

// 原searchService.search方法
search() {
  const apiCall1$ = this.http.get('/api/search1');
  const apiCall2$ = this.http.get('/api/search2');
  return forkJoin([apiCall1$, apiCall2$]);
}

// 调用处
this.searchService.search().subscribe(([result1, result2]) => {
  this.store.dispatch(new SearchSuccess(result1));
  this.store.dispatch(new SearchSuccess(result2));
});

现在改成用merge来实现：

// 修改后的searchService.search方法
search() {
  const apiCall1$ = this.http.get('/api/search1');
  const apiCall2$ = this.http.get('/api/search2');
  // 用merge合并两个Observable，每个结果会单独发出
  return merge(apiCall1$, apiCall2$);
}

// 调用处，每个响应回来就立即触发动作
this.searchService.search().subscribe((singleResult) => {
  this.store.dispatch(new SearchSuccess(singleResult));
});

额外优化：区分不同请求的结果 #

如果需要知道当前响应来自哪个API（比如后续处理有差异化逻辑），可以给每个请求的结果加上标识：

search() {
  const api1WithMeta$ = this.http.get('/api/search1').pipe(
    map(data => ({ source: 'searchApi1', payload: data }))
  );
  const api2WithMeta$ = this.http.get('/api/search2').pipe(
    map(data => ({ source: 'searchApi2', payload: data }))
  );
  
  return merge(api1WithMeta$, api2WithMeta$);
}

// 调用时可以区分来源
this.searchService.search().subscribe((result) => {
  console.log(`处理来自${result.source}的响应`);
  this.store.dispatch(new SearchSuccess(result.payload));
});

为什么不用其他操作符？ #

• concat：会按顺序发起请求，等第一个完成再发第二个，不符合你原来并行执行的需求。
• race：只会取第一个完成的请求结果，直接忽略其他请求，显然不适用。
• forkJoin：就是你现在用的，必须等全部完成才返回，不符合即时处理的要求。

这样改造后，两个请求依然是并行发起的，但每个响应一回来就会立即触发SearchSuccess动作，完全满足你的需求。

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