如何限制HTTP并行请求？实现分批4个请求循环执行方案

原代码使用forkJoin会一次性发起所有分片上传请求，当分片数量较多时容易引发请求过载问题。要实现每次并行发起4个请求，完成整批后再启动下一批，可以通过RxJS的from、bufferCount和concatMap组合实现，以下是修改后的代码：

// 原有的分片生成逻辑保持不变
for (let offset = 0; offset < file[0].size; offset += chunkSize) {
  const chunk = file[0].slice(offset, offset + chunkSize);
  
  const blockId = String(Math.floor(100000 + Math.random() * 900000));
  const base64BlockId = btoa(blockId);

  this.blockids.push(base64BlockId);

  request.push(this.blobService.addNewVideo(uniqueName, chunk, base64BlockId));
}

// 分批处理请求，每次并行4个
from(request)
  .pipe(
    // 将请求数组按每4个一组拆分
    bufferCount(4),
    // 依次处理每组，只有当前组全部完成才会进入下一组
    concatMap(group => 
      forkJoin(
        group.map(req => req.pipe(
          tap((event: HttpEvent<any>) => {
            switch (event.type) {
              case HttpEventType.UploadProgress:
                // 注意：此处原逻辑会被多个请求覆盖进度，可改为累计计算总进度
                this.progress = Math.round((event.loaded / event.total!) * 100);
                break;
              case HttpEventType.Response:
                console.log('Successfully uploaded', event.body," : ",event.status,event.statusText);
            }
          })
        ))
      )
    ),
    // 等待所有批次处理完成
    last()
  )
  .subscribe(() => {
    this.commit(uniqueName);
  });

关键逻辑说明 #

• from(request): 将请求数组转换为Observable序列，逐个发出每个请求Observable
• bufferCount(4): 把序列中的请求按每4个打包成一个数组，形成"批次"
• concatMap(group => forkJoin(group)): 对每个批次使用forkJoin并行执行其中的4个请求，concatMap保证前一个批次全部完成后才会处理下一个批次
• last(): 忽略中间批次的结果，只在所有批次处理完成后触发后续的commit操作

补充：非严格分批的并发限制 #

如果不需要严格等待整批完成再启动下一批，而是希望同时保持最多4个请求在运行（完成一个就立即补充下一个），可以改用mergeMap并指定并发数：

from(request)
  .pipe(
    mergeMap(req => req.pipe(
      tap((event: HttpEvent<any>) => {
        // 进度处理逻辑同上
        switch (event.type) {
          case HttpEventType.UploadProgress:
            this.progress = Math.round((event.loaded / event.total!) * 100);
            break;
          case HttpEventType.Response:
            console.log('Successfully uploaded', event.body," : ",event.status,event.statusText);
        }
      })
    ), 4), // 此处指定并发数为4
    last()
  )
  .subscribe(() => {
    this.commit(uniqueName);
  });

进度计算优化提示 #

原代码中this.progress会被多个上传请求的进度事件覆盖，导致进度显示不准确。可以通过累计已上传的总字节数，结合文件总大小计算真实的整体进度：

let totalUploaded = 0;
const totalSize = file[0].size;

// 在tap的UploadProgress分支中修改为：
if (event.total) {
  totalUploaded += event.loaded - (event.total - event.loaded); // 累计当前分片已上传字节
  this.progress = Math.round((totalUploaded / totalSize) * 100);
}

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