如何利用RxJS批量发送携带不同索引的HTTP请求？

我需要批量发送多个携带不同索引的HTTP请求：现有documentIndexes数组，要将每个索引作为indx参数传入getData()方法，示例请求如下：

this.document = await firstValueFrom(this._dataService.getData(
options.Id,
options.rawIndex,
indx (1)
));
this.document = await firstValueFrom(this._dataService.getData(
options.Id,
options.rawIndex,
indx (2)
));
this.document = await firstValueFrom(this._dataService.getData(
options.Id,
options.rawIndex,
indx (3)
));

目前用for循环实现功能正常，但想了解基于RxJS的更优实现方案，当前代码如下：

const documentIndexes = [1,2,3]

for (let index = 0; index < documentIndexes.length; index++) {
      const indx = documentIndexes[index];
      this.document = await firstValueFrom(this._dataService.getData(
        options.Id,
        options.rawIndex,
        indx
      ));
      this.allDoc.push(this.document)
}

1. 并行请求（效率最高，无顺序要求） #

如果不需要严格按索引顺序发起请求，用forkJoin一次性触发所有请求，等全部完成后统一处理结果：

import { forkJoin } from 'rxjs';

const documentIndexes = [1,2,3];

// 将每个索引转换为对应的Observable请求
const requestObservables = documentIndexes.map(indx => 
  this._dataService.getData(options.Id, options.rawIndex, indx)
);

// 订阅所有请求，全部完成后拿到按原数组顺序排列的结果
forkJoin(requestObservables).subscribe({
  next: (documents) => {
    this.allDoc.push(...documents);
  },
  error: (err) => {
    console.error('请求出错:', err);
  }
});

// 也可转为async/await风格
// const documents = await firstValueFrom(forkJoin(requestObservables));
// this.allDoc.push(...documents);

2. 顺序请求（严格按数组顺序执行） #

如果必须等待前一个请求完成后再发起下一个，用concatMap实现串行请求：

import { from, concatMap } from 'rxjs';

const documentIndexes = [1,2,3];

from(documentIndexes).pipe(
  concatMap(indx => this._dataService.getData(options.Id, options.rawIndex, indx))
).subscribe({
  next: (document) => {
    this.allDoc.push(document);
  },
  error: (err) => {
    console.error('请求出错:', err);
  },
  complete: () => {
    console.log('所有请求完成');
  }
});

3. 带并发限制的并行请求 #

如果数组过大，不想一次性发起过多请求（避免后端限流），用mergeMap设置并发数：

import { from, mergeMap } from 'rxjs';

const documentIndexes = [1,2,3];
const concurrency = 2; // 同时最多发起2个请求

from(documentIndexes).pipe(
  mergeMap(indx => this._dataService.getData(options.Id, options.rawIndex, indx), concurrency)
).subscribe({
  next: (document) => {
    this.allDoc.push(document);
  },
  error: (err) => {
    console.error('请求出错:', err);
  }
});

• forkJoin：适合无顺序要求的批量请求，性能最优；仅当所有请求成功时返回结果，任一请求失败则触发整体错误
• concatMap：适合必须串行执行的场景，性能稍差，但能严格保证请求顺序
• mergeMap（带并发）：适合大数组场景，平衡请求性能和后端压力

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