NgRx Effect中循环调用API并派发进度Action的实现问题

你写的原始代码存在4个核心问题：

• 嵌套combineLatest会等待所有预约的所有订单请求全部完成后才一次性输出结果，无法实现每完成一个预约的更新就实时派发进度Action的需求
• catchError中返回of(console.info(message))会将非Action类型的值注入Effect流，NgRx Effect要求流中所有输出值都是合法的Action实例，类型不匹配直接触发运行时报错
• 两层combineLatest返回的是二维数组结构的流输出，没有经过打平处理，NgRx无法识别为可派发的Action
• 进度计数逻辑不符合需求：你当前是每完成一个订单就触发进度更新，但需求中numberOfReservationsUpdated统计的是已完成更新的预约数，应该是单个预约下的所有订单全部更新完成后，才计数+1并派发进度Action

你需要补充from、forkJoin、concatMap（或mergeMap，根据并发需求选择）、map操作符，修复后代码如下：

@Effect()
updateOrders$ = this.actions$.pipe(
  ofType<UpdateOrders>(UpdateOrdersActionType.UPDATE_ORDERS),
  filter((action) => !!action.reservationOrders),
  exhaustMap((action) => {
    const totalReservations = action.reservationOrders.length;
    // 将预约数组转为逐个emit的Observable流
    return from(action.reservationOrders).pipe(
      // concatMap保证预约按顺序逐个处理，需要并行处理可替换为mergeMap
      concatMap((reservation, index) => {
        // 等待当前预约下所有订单更新请求全部完成，才算该预约更新完成
        return forkJoin(
          reservation.orders.map(order => {
            const orderUpdateRequest: OrderUpdateRequest = {
              orderId: order.orderId,
              amount: order.amount
            };
            return this.orderApiService.updateOrder(orderUpdateRequest).pipe(
              // 单个订单请求错误不阻塞整个流程，可根据业务需求调整错误处理逻辑
              catchError((err) => {
                console.info(err);
                // 错误场景也返回空值让forkJoin继续执行，避免整个预约更新中断
                return of(null);
              })
            )
          })
        ).pipe(
          // 单个预约全部订单处理完成后，派发进度Action
          map(() => new UpdateOrderProgress(index + 1, totalReservations))
        )
      })
    )
  })
);

如果你的业务要求单个预约下所有订单全部更新成功才算该预约更新完成，可以去掉updateOrder内层的catchError，在forkJoin外层统一加catchError处理即可。

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