RxJS按固定间隔处理数组每项调用2个服务并跟踪处理状态的实现方案

实现方案 #

核心逻辑分为3部分：

1. 给每个id对应的forkJoin聚合请求添加错误捕获，避免单个请求失败中断整个处理链路，同时返回结构化结果，统一携带id、执行状态、业务数据/错误信息
2. 用concatMap + timer实现串行间隔执行，保证每10秒才处理下一个id的请求，不会出现请求堆叠压垮服务的问题
3. 在回调中聚合区分成功、失败的id列表

依赖导入

import { forkJoin, from, timer, of } from 'rxjs';
import { concatMap, map, catchError } from 'rxjs/operators';

完整业务代码

const arraytoprocess = [...this.selection.selected]; // 待处理id数组
const successList = []; // 存储处理成功的id及返回结果
const failList = []; // 存储处理失败的id及错误信息

from(arraytoprocess).pipe(
  // 间隔控制：第一个请求立即执行，后续每个请求间隔10秒，如需第一个也延迟10秒可直接写10000
  concatMap((item, index) => timer(index === 0 ? 0 : 10000).pipe(
    // 执行当前id的两个服务聚合请求
    concatMap(() => forkJoin([
      this._api.service1(+item.id),
      this._api.service2(+item.id)
    ]).pipe(
      // 成功时包装返回结构
      map(([service1Res, service2Res]) => ({
        id: item.id,
        success: true,
        service1Res,
        service2Res
      })),
      // 捕获当前id的请求错误，转换为正常返回值，不会中断后续其他id的处理
      catchError(err => of({
        id: item.id,
        success: false,
        error: err
      }))
    ))
  ))
).subscribe(
  res => {
    // 按状态分类存储结果
    res.success ? successList.push(res) : failList.push(res);
  },
  globalErr => {
    // 全局异常兜底，正常逻辑不会触发，仅用于捕获未预料的错误
    console.error('全局处理异常：', globalErr);
  },
  () => {
    // 所有id处理完成回调，可在这里统一处理最终结果
    console.log('全部处理完成，成功列表：', successList, '失败列表：', failList);
  }
)

注意说明

• concatMap会严格保证执行顺序：前一个id的两个服务请求全部完成后，才会开始计时等待下一个10秒，执行后续id的请求
• 单个id的两个服务只要任意一个报错，都会被归到失败列表，不会影响其他id的正常处理
• 如果需要调整间隔时间，直接修改timer的延迟参数即可

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