NgRx中并发处理Observable数组与条件HTTP请求的优化问询

我有一个NgRx选择器data$: Observable<Data[]>，值为对象数组。组件中需要遍历每个数据项发起两次HTTP请求，其中一次是条件请求（取决于另一个ObservableisConditionTrue$: Observable<boolean>的值）。数据通过async管道在表格中订阅展示，额外HTTP请求会映射data的部分属性，要求满足以下执行逻辑：

• 组件初始化时，data$在表格中订阅，值为空数组[]
• data$首次发射长度为5的数组，表格立即展示5条数据
• 遍历这5条数据，每条触发httpCallA
• 若isConditionTrue$返回true，部分数据触发对应类型的条件HTTP请求
• 任一请求完成后，表格重新渲染对应条目，且部分请求失败不中断其他请求执行

我尝试过将请求结果存入数组但产生冗余数据，也考虑过通过ID匹配数据，但现有代码复杂可读性差，寻求优化方案。

现有代码 #

vm$ = combineLatest([data$, isConditionTrue$]).pipe(
  switchMap(([data, isConditionTrue]) => {
    // no data
    if (!data?.length) {
      return of({ data: [], isConditionTrue });
    }
    const result = [];
    // loop through data
    return from(data).pipe(
      // create concurent HTTP calls
      mergeMap((dataItem) => {
        // combine multiple http request
        return merge(
          // first http call (should be performed everytime)
          httpCallA(dataItem.id).pipe(
            map((response) => {
              if (!response) {
                return dataItem;
              }
              return {
                ...response,
                ...dataItem,
              };
            }),
            catchError((error) => {
              console.error(error);
              return of(dataItem);
            }),
          ),
          // second http request (conditional)
          of(dataItem).pipe(
            filter(() => isConditionTrue), // only perform when condition is true
            mergeMap(() => {
              // if dataItem is of specific type call HTTP call
              if (dataItem.conditionA) {
                return httpCallB(dataItem.id).pipe(
                  map((res) => ({ ...res, dataItem })),
                  catchError((error) => {
                    console.error(error);
                    return of(dataItem);
                  }),
                );
              }

              // otherwise
              return httpCallC(dataItem.id).pipe(
                map((res) => ({ ...res, dataItem })),
                catchError((error) => {
                  console.error(error);
                  return of(dataItem);
                }),
              );
            }),
          ),
        );
      }),
      tap((value) => result.push(value)),
      map(() => ({ data: result, isConditionTrue })),
    );
  }),
  startWith({ data: [], isConditionTrue: false }),
);

优化方案 #

核心思路 #

1. 用ID跟踪数据项，避免冗余数据，每次请求完成后只更新对应ID的数据
2. 拆分请求逻辑，把单个数据项的请求处理封装成独立函数，提升可读性
3. 保持并发执行，使用mergeMap控制请求并发
4. 优先返回原始数据，确保表格立即展示，再逐步更新请求结果

优化后的代码 #

// 封装单个数据项的请求处理逻辑
const processDataItem = (dataItem: Data, isConditionTrue: boolean) => {
  // 基础请求：每次必发
  const baseRequest$ = httpCallA(dataItem.id).pipe(
    map(res => res ? { ...dataItem, ...res } : dataItem),
    catchError(err => {
      console.error('httpCallA失败:', err);
      return of(dataItem);
    })
  );

  // 条件请求：仅当isConditionTrue为true时执行
  const conditionalRequest$ = isConditionTrue ? (
    dataItem.conditionA ? httpCallB(dataItem.id) : httpCallC(dataItem.id)
  ).pipe(
    map(res => res ? { ...dataItem, ...res } : dataItem),
    catchError(err => {
      console.error(`条件请求失败:`, err);
      return of(dataItem);
    })
  ) : of(null); // 条件不满足时返回null，后续过滤掉

  // 合并两个请求，过滤null值
  return merge(baseRequest$, conditionalRequest$).pipe(
    filter(Boolean)
  );
};

// 构建视图模型Observable
vm$ = combineLatest([data$, isConditionTrue$]).pipe(
  switchMap(([data, isConditionTrue]) => {
    if (!data?.length) {
      return of({ data: [], isConditionTrue });
    }

    // 初始先返回原始数据，确保表格立即展示
    const initialData$ = of({ data: [...data], isConditionTrue });

    // 处理所有数据项的请求，收集更新后的条目
    const updates$ = from(data).pipe(
      mergeMap(item => processDataItem(item, isConditionTrue)),
      // 用scan维护最新的数据数组：根据ID匹配更新
      scan((acc, updatedItem) => 
        acc.map(item => item.id === updatedItem.id ? updatedItem : item),
        [...data]
      ),
      map(updatedData => ({ data: updatedData, isConditionTrue }))
    );

    // 合并初始数据和后续更新，确保先展示原始数据再逐步更新
    return merge(initialData$, updates$);
  }),
  startWith({ data: [], isConditionTrue: false })
);

优化点说明 #

• 数据更新逻辑：使用scan操作符维护数据数组，每次请求完成后通过ID匹配替换对应条目，彻底避免冗余数据
• 请求逻辑封装：把单个数据项的请求处理抽成独立函数processDataItem，代码结构更模块化，可读性大幅提升
• 初始数据优先展示：先发射原始数据，保证表格立即渲染初始内容，后续请求完成后再逐步更新对应条目
• 独立错误处理：每个HTTP请求单独捕获错误，某一个请求失败不会中断其他请求的执行
• 条件请求逻辑简化：用三元表达式替代嵌套管道，条件判断逻辑更直观易懂

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