如何使用RxJS的merge()与from()合并多请求响应

嘿，我来给你捋捋怎么用RxJS的merge()和from()处理这种多HTTP请求的场景！

先确认下我get到你的需求了：你得发起n个（数量由用户选择的日期范围决定）HTTP请求，每个月的数据单独查询，URL都是根据用户输入自动生成的对吧？比如查1年数据的话，URL数组大概是这样：

[
  "https://opendata.nhsbsa.net/api/3/action/datastore_search?resource_id=EPD_202301&sql=SELECT * from `EPD_202301` WHERE BNF_CODE = '0410030C0AAAFAF'",
  // 这里省略另外11个对应2023年其他月份的URL
]

下面我给你一步步拆解实现方式，都是实打实的RxJS操作：

第一步：把单个请求转成Observable #

首先得把HTTP请求包装成RxJS的Observable，毕竟RxJS的操作都是基于流的。因为fetch（你也可以用Axios这类库）返回的是Promise，from()刚好能把Promise转成Observable，完美适配RxJS的流操作。

import { from } from 'rxjs';
import { switchMap, catchError } from 'rxjs/operators';

// 封装请求函数，输入URL返回请求Observable
const fetchMonthData = (url: string) => {
  return from(fetch(url)).pipe(
    // 把fetch响应转成JSON格式
    switchMap(response => response.json()),
    // 可选：给单个请求加错误处理，避免一个请求挂了搞崩整个流
    catchError((err) => {
      console.error(`请求 ${url} 失败:`, err);
      return from([null]); // 返回空值或默认值，让其他请求继续执行
    })
  );
};

第二步：用merge()合并所有请求Observable #

接下来把你的URL数组批量转成请求Observable，再用merge()把它们合并成一个流。merge()会并行发起所有请求，不管哪个请求先完成，它的响应都会立刻被推出来。

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

// 假设这是你生成好的URL数组
const requestUrls = [
  "https://opendata.nhsbsa.net/api/3/action/datastore_search?resource_id=EPD_202301&sql=SELECT * from `EPD_202301` WHERE BNF_CODE = '0410030C0AAAFAF'",
  "https://opendata.nhsbsa.net/api/3/action/datastore_search?resource_id=EPD_202302&sql=SELECT * from `EPD_202302` WHERE BNF_CODE = '0410030C0AAAFAF'",
  // ...剩下月份的URL
];

// 把每个URL转成请求Observable，再合并成一个流
const mergedRequest$ = merge(...requestUrls.map(url => fetchMonthData(url)));

第三步：订阅流处理响应 #

现在你可以订阅这个合并后的流，有两种常见的处理方式：

1. 逐个处理响应：每个请求完成就立刻处理它的结果

mergedRequest$.subscribe({
  next: (monthData) => {
    if (monthData) {
      console.log('拿到一个月的数据啦:', monthData);
      // 这里可以做单月数据的即时处理，比如渲染到页面、存临时变量
    }
  },
  complete: () => {
    console.log('所有月份的请求都搞定啦！');
  }
});

2. 收集所有响应统一处理：如果需要等所有请求完成后，拿到包含所有响应的数组再批量处理，加个toArray()操作符就行

mergedRequest$.pipe(toArray()).subscribe({
  next: (allMonthData) => {
    console.log('所有月份的数据都齐了:', allMonthData);
    // 这里可以做批量操作，比如合并数据、统计分析
  },
  complete: () => {
    console.log('全部请求执行完毕');
  }
});

一些实用小提醒 #

• 如果请求数量特别多（比如几十上百个），并行发起可能触发服务器的频率限制，这时候可以换成mergeMap并设置并发数，或者用concat串行发起请求。
• 一定要给单个请求加catchError，不然某个请求失败会直接终止整个合并流，其他请求的结果也拿不到了。

备注：内容来源于stack exchange，提问作者Kiran