Angular与RxJS：如何扩展forkJoin实现多HTTP请求合并并即时输出

问题描述 #

现有组件代码通过RxJS的switchMap和forkJoin为每个订单补充客户名称：

extendedOrders$ = this.salesService.getOrders().pipe(
  switchMap(orders => {
    return forkJoin(orders.map(order => {
      return this.salesService.getOrderCustomers(order.id).pipe(
        map(customer => ({ ...order, customerName: customer.name })),
      );
    }));
  }),
);

想扩展这段代码，在forkJoin内部再发起另一个getById类的服务请求，将响应结果也合并到同一个订单对象中。

更新：尝试的代码 #

自行编写的代码如下，但未达到预期效果：

this.salesService.getOrders().pipe(
  switchMap(orders => {
    return forkJoin(
      orders.map(order => {
        return {
          ...order,
          idAttr1: this.service1.method1(order.idAttr1).pipe(
            map(result => result.name)
          ),
          idAttr2: this.service2.method2(order.idAttr2).pipe(
            map(result => result.name)
          ),
        };
      })
    );
  })
)
.subscribe((result) => {
  console.log(result);
});

解决方案 #

当前写法的问题在于：直接将Observable对象赋值给了订单属性，而非等待请求完成后获取实际返回值。forkJoin需要接收Observable数组，而非包含Observable属性的普通对象数组。

正确的实现方式是，对每个订单用内层forkJoin并行发起多个API请求，待所有请求完成后再将结果合并到原订单对象：

extendedOrders$ = this.salesService.getOrders().pipe(
  switchMap(orders => {
    // 为每个订单创建处理Observable：并行获取额外数据后合并到订单
    const orderObservables = orders.map(order => {
      return forkJoin({
        // 获取第一个请求的名称字段
        attr1Name: this.service1.method1(order.idAttr1).pipe(map(res => res.name)),
        // 获取第二个请求的名称字段
        attr2Name: this.service2.method2(order.idAttr2).pipe(map(res => res.name))
      }).pipe(
        // 将请求结果合并到原订单对象
        map(({ attr1Name, attr2Name }) => ({
          ...order,
          idAttr1: attr1Name,
          idAttr2: attr2Name
        }))
      );
    });

    // 等待所有订单的处理流程完成
    return forkJoin(orderObservables);
  })
);

代码说明 #

1. 内层forkJoin({ ... })用于并行发起多个API请求，相比串行请求能提升处理效率。
2. 内层请求完成后，通过map操作符将返回的字段值合并到原订单对象，生成包含完整数据的订单Observable。
3. 外层将所有订单的Observable收集为数组，再通过forkJoin等待所有订单处理完成，最终得到补充完所有数据的订单数组。

如果需要保留原有的客户名称补充逻辑，只需将getOrderCustomers请求加入内层forkJoin即可：

extendedOrders$ = this.salesService.getOrders().pipe(
  switchMap(orders => {
    const orderObservables = orders.map(order => {
      return forkJoin({
        customer: this.salesService.getOrderCustomers(order.id),
        attr1Name: this.service1.method1(order.idAttr1).pipe(map(res => res.name)),
        attr2Name: this.service2.method2(order.idAttr2).pipe(map(res => res.name))
      }).pipe(
        map(({ customer, attr1Name, attr2Name }) => ({
          ...order,
          customerName: customer.name,
          idAttr1: attr1Name,
          idAttr2: attr2Name
        }))
      );
    });

    return forkJoin(orderObservables);
  })
);

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