Angular/Ionic用forkJoin分批调API 全部完成后统一触发订阅

核心逻辑拆分：

• 将原始请求数组按每10个一组切分为多个批次
• 批次间串行执行，上一批次所有请求返回后等待1000ms再发起下一批
• 单批次内部用forkJoin并行发起请求，保证单批次请求效率
• 所有批次执行完成后，将结果按原请求顺序拼接为一维数组，一次性触发订阅回调

完整实现代码 #

首先引入需要的RxJS操作符：

import { forkJoin, from } from 'rxjs'
import { concatMap, delay, map, toArray } from 'rxjs/operators'

实现分批逻辑和请求方法：

/**
 * 按指定大小将数组拆分为多批
 * @param list 原始数组
 * @param batchSize 单批大小
 */
const splitToBatches = <T>(list: T[], batchSize = 10): T[][] => {
  const batches: T[][] = []
  for (let i = 0; i < list.length; i += batchSize) {
    batches.push(list.slice(i, i + batchSize))
  }
  return batches
}

public demoCall() {
  // 构造原始请求列表，替换为实际接口逻辑即可
  const reqList = new Array(45).fill(null).map((_, idx) => {
    return this.http.post('/your/api/url', { data: idx + 1 })
  })

  // 拆分批次，配置批次间隔时间
  const batches = splitToBatches(reqList, 10)
  const BATCH_DELAY = 1000

  return from(batches).pipe(
    // 串行处理每个批次，保证批次执行顺序
    concatMap((singleBatch, batchIndex) => {
      // 单批次内并行发起所有请求
      return forkJoin(singleBatch).pipe(
        // 最后一批请求完成后不需要额外等待
        delay(batchIndex === batches.length - 1 ? 0 : BATCH_DELAY)
      )
    }),
    // 收集所有批次的返回结果为二维数组
    toArray(),
    // 拍平为一维数组，保持和原始请求一致的顺序
    map(allBatchResult => allBatchResult.flat())
  )
}

调用方式 #

和预期调用形式完全一致，所有请求完成后订阅回调仅触发一次：

this.demoCall().subscribe(res => {
  // res为长度45的结果数组，顺序和req1~req45一一对应
  console.log('全部请求执行完成', res)
})

注意事项 #

• 单批次内10个请求为并行发起，不会额外拖慢整体请求速度
• 任意一个请求报错时，整个流会抛出错误，和原生forkJoin行为一致；如果需要容忍单个请求失败，可以在单个请求后挂载catchError返回兜底值
• 最终返回的结果顺序和原始请求数组顺序完全一致，不需要额外排序处理

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