高并发请求场景下使用forkJoin分块处理如何替代for循环？

RxJS 原生并发控制方案（推荐） #

无需手动拆分请求数组、无需编写循环遍历逻辑，直接通过mergeMap的内置并发参数实现限流，效果和你当前手动分块调用forkJoin完全一致，且请求返回顺序和原始请求数组顺序匹配：

import { from, lastValueFrom, of } from 'rxjs';
import { mergeMap, toArray, catchError } from 'rxjs/operators';

// 所有请求的Observable数组，和你之前传给forkJoin的数组一致
const requestList = [...]; 
// 并发数设置为10，和你之前的分块大小匹配，可按需调整
const CONCURRENCY_LIMIT = 10;

// 转Promise的写法（RxJS 7+ 推荐使用lastValueFrom替代toPromise）
const fetchAllData = async () => {
  try {
    return await lastValueFrom(
      from(requestList).pipe(
        // 第二个参数为并发数，最多同时执行10个请求，跑完自动取下一个
        mergeMap(request => request, CONCURRENCY_LIMIT),
        // 收集所有返回结果为数组，和forkJoin返回格式一致
        toArray()
      )
    );
  } catch (err) {
    // 全局错误处理逻辑
    console.error('请求出错：', err);
    throw err;
  }
};

如果需要单独处理单个请求的错误，避免单个请求失败导致全部结果丢失，可以在mergeMap内给单个请求加错误捕获：

mergeMap(request => request.pipe(catchError(err => of({ error: err }))), CONCURRENCY_LIMIT)

纯Promise无循环方案 #

如果你不想依赖RxJS的操作符，也可以通过Promise.all结合分组控制实现，无需手动编写循环遍历分块的逻辑：

const CONCURRENCY_LIMIT = 10;
const requestList = [...]; // 所有请求的Promise创建函数数组

const fetchAllData = async () => {
  const finishedGroups = new Set();
  return Promise.all(
    requestList.map(async (requestFn, index) => {
      const groupIdx = Math.floor(index / CONCURRENCY_LIMIT);
      // 等待前一组所有请求完成
      if (groupIdx > 0) {
        while (!finishedGroups.has(groupIdx - 1)) {
          await new Promise(resolve => setTimeout(resolve, 20));
        }
      }
      const result = await requestFn();
      // 标记当前组完成
      if (index % CONCURRENCY_LIMIT === CONCURRENCY_LIMIT - 1) {
        finishedGroups.add(groupIdx);
      }
      return result;
    })
  );
};

方案对比 #

• RxJS方案代码更简洁，内置并发控制逻辑无额外冗余代码，调整并发数仅需修改参数，更适合现有基于RxJS的项目
• 纯Promise方案无第三方依赖，适合不使用RxJS的场景

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