Angular RxJS 并行调用多接口 按响应到达顺序处理赋值

你当前使用的forkJoin、zip操作符的设计逻辑是等待所有上游Observable执行完成后，才将聚合结果统一推送，无法满足单条响应到达就立刻处理的需求。这类多源响应实时处理的场景，适合使用RxJS的merge操作符实现多请求流的合并。

merge操作符的匹配特性 #

• 可将多个HTTP请求对应的Observable输出，按照响应实际到达的时间顺序推入统一处理流
• 任意一个请求返回有效响应时，处理流会立刻推送该响应，不会等待其他请求执行完成
• 控制台打印顺序完全匹配网络响应的真实到达顺序，不会固定源请求的预设顺序
• 因为你在每个服务方法中已经内置了catchError错误捕获逻辑，单个请求失败不会导致整个合并流中断，失败请求的返回值会和其他成功响应一样按到达顺序被处理

修正后的组件代码 #

首先修正你原有代码里的两处笔误：

1. 组件内定义请求时，req2、req3重复调用了GetDataServer1方法，会导致三个请求全部指向同一台服务器，无法测试不同服务器的响应差异
2. 原forkJoin的入参数组中重复传入了req2，漏写了req3

完整实现步骤：

1. 导入需要的RxJS工具：

   import { merge } from 'rxjs';
   import { map } from 'rxjs/operators';
   

2. 调整组件内的getData方法逻辑：

   getData(data: any) {
     // 定义三个对应不同服务器的请求流，增加来源标记方便后续区分赋值
     const req1$ = this.service.GetDataServer1(data).pipe(
       map(res => ({ source: 'server1', data: res }))
     );
     const req2$ = this.service.GetDataServer2(data).pipe(
       map(res => ({ source: 'server2', data: res }))
     );
     const req3$ = this.service.GetDataServer3(data).pipe(
       map(res => ({ source: 'server3', data: res }))
     );
   
     // 合并三个流，按实际到达顺序处理每个响应
     merge(req1$, req2$, req3$).subscribe(markedRes => {
       // 打印顺序完全和响应的实际到达顺序一致
       console.log(markedRes.data);
       // 根据来源标记给对应变量赋值
       switch (markedRes.source) {
         case 'server1':
           this.server1List = markedRes.data;
           break;
         case 'server2':
           this.server2List = markedRes.data;
           break;
         case 'server3':
           this.server3List = markedRes.data;
           break;
       }
     });
   }
   

常见易混操作符的区别（避免踩坑） #

• forkJoin：等待所有请求全部完成后一次性返回所有结果的数组，适合需要全量数据再统一处理的场景，不适合实时单条响应处理
• zip：按照上游流的推送顺序一一配对值，等每个位置的流都推送值后才返回聚合数组，不匹配单次HTTP请求类Observable的处理逻辑
• race：仅保留所有请求中第一个返回的有效响应，直接丢弃后续其他请求的结果，不符合你需要接收所有服务器响应的需求
• merge：无等待合并所有上游流的推送值，是当前场景下的适配实现

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