如何在Angular NGRX Effect中拆分大数据组为块并依次请求API后合并结果

嘿，我刚好处理过类似的NGRX批量请求场景，给你梳理下解决方案和修改后的代码！

首先，我们需要解决三个核心问题：拆分大数组、按顺序发送批量请求、累积所有响应结果。这里用RxJS的操作符就能完美实现，不需要转Promise哈。

第一步：写一个数组拆分工具函数 #

先实现一个简单的函数，把你的大数组按每100个对象拆分成小块：

// 通用数组拆分函数，支持任意类型的数组
const chunkArray = <T>(array: T[], chunkSize: number): T[][] => {
  const chunks = [];
  for (let i = 0; i < array.length; i += chunkSize) {
    chunks.push(array.slice(i, i + chunkSize));
  }
  return chunks;
};

第二步：修改你的Effect逻辑 #

接下来调整你的loadRows$ Effect，替换原来的请求逻辑，加入拆分、顺序请求和结果合并的逻辑：

loadRows$ = createEffect(() => {
  return this.actions$.pipe(
    ofType(RowsActions.LoadRows),
    switchMap((action) => {
      // 1. 将传入的大数组拆分为每100个一组的小块
      const rowChunks = chunkArray(action.rows, 100);

      // 2. 将块数组转为Observable流，逐个处理每个块
      return from(rowChunks).pipe(
        // 使用concatMap保证请求**按顺序发送**（前一个请求完成后再发下一个）
        // 如果你的API支持并行请求，可以换成mergeMap，但要注意控制并发数
        concatMap((chunk) => this.eventsService.getBatchRows(chunk)),
        // 3. 累积每个请求的响应结果
        scan((accumulatedRows, currentResponse) => {
          // 把当前请求返回的rows合并到之前的累积数组中
          return [...accumulatedRows, ...currentResponse.rows];
        }, [] as YourRowType[]), // 替换成你实际的行数据类型
        // 4. 只取最后一次累积的结果（所有请求完成后的最终合并数组）
        last(),
        // 5. 分发成功的Actions，更新Store
        switchMap((mergedRows) => {
          return [
            RowsActions.RowsLoaded({ rows: mergedRows }),
            RowsActions.LoadedTableData(mergedRows),
          ];
        }),
        // 6. 全局错误处理：任何一个请求失败都会触发这里
        catchError(() => {
          return of(RowsActions.RowsFailedToLoad());
        })
      );
    })
  );
});

关键操作符解释 #

• from(rowChunks)：把拆分后的块数组转换成Observable流，每次发射一个小块。
• concatMap：强制请求按顺序执行，避免同时发送大量请求压垮API。如果你的API支持并行，可以换成mergeMap并搭配并发参数（比如mergeMap(chunk => ..., 3)表示同时最多发3个请求）。
• scan：相当于RxJS版本的Array.reduce，每次新请求返回结果时，就把它合并到之前的累积数组里，这样我们能实时跟踪合并进度。
• last()：因为scan会在每次累积后都发射值，我们只需要最后一次（所有请求完成后的最终合并结果），所以用last()来过滤出这个最终值。

可选：容错处理 #

如果希望某个块请求失败时，不中断整个批量流程（比如跳过失败的块，继续处理其他块），可以把catchError放到concatMap内部：

concatMap((chunk) => 
  this.eventsService.getBatchRows(chunk).pipe(
    // 单个块请求失败时，返回空数组，不中断整个流
    catchError((error) => {
      console.error(`处理块失败:`, error);
      return of({ rows: [] });
    })
  )
)

这样即使某个块请求失败，其他块依然会正常处理，最终合并的结果会包含所有成功请求的行数据。

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