Angular中使用forkJoin合并多个Observable订阅结果问题

问题本质 #

• 原代码每次订阅getEvaluatorSaaSRequests时直接覆盖数组，导致最终仅保留最后一个请求的结果。
• 临时push方案会重复添加数据，因为snapshotChanges是持续监听Firestore数据变更的Observable，每次数据更新都会触发回调，造成重复。

正确实现方案 #

利用RxJS的forkJoin并行处理多个Observable，配合操作符整合数据流，同时避免重复数据问题：

组件核心代码 #

import { forkJoin, from } from 'rxjs';
import { map, switchMap, take } from 'rxjs/operators';

demandesSaaSEvaluator: Demande[] = [];

// 将Firestore的Promise转为Observable，纳入RxJS流处理
from(this.fire.collection('businessUnit')
  .where('evaluators', 'array-contains', this.auth.currentUserEmail).get())
.pipe(
  // 提取所有匹配的nomBu列表
  map(snapshot => snapshot.docs.map(doc => doc.data().nomBu)),
  // 并行请求每个nomBu对应的SaaS数据
  switchMap(nomBuList => {
    // 为每个nomBu创建带终止条件的Observable
    const requestObservables = nomBuList.map(nomBu => 
      this.service.getEvaluatorSaaSRequests(nomBu).pipe(
        take(1), // 仅获取一次数据，避免持续监听导致重复
        map(actionArray => actionArray.map(item => ({
          id: item.payload.doc.id,
          ...(item.payload.doc.data() as object)
        } as Demande)))
      )
    );
    // 合并所有Observable的结果
    return forkJoin(requestObservables);
  })
)
.subscribe(allResults => {
  // 将二维结果数组合并为一维数组
  this.demandesSaaSEvaluator = allResults.flat();
  // 可选：按id去重（若不同nomBu存在相同id的数据）
  // this.demandesSaaSEvaluator = [...new Map(allResults.flat().map(item => [item.id, item])).values()];
});

实时更新场景适配 #

如果需要实时监听Firestore数据变化，改用combineLatest并添加去重逻辑：

import { combineLatest, from } from 'rxjs';
import { map, switchMap } from 'rxjs/operators';

// ... 其他代码保持一致
switchMap(nomBuList => {
  const requestObservables = nomBuList.map(nomBu => 
    this.service.getEvaluatorSaaSRequests(nomBu).pipe(
      map(actionArray => actionArray.map(item => ({
        id: item.payload.doc.id,
        ...(item.payload.doc.data() as object)
      } as Demande)))
    )
  );
  return combineLatest(requestObservables);
})
.subscribe(allResults => {
  const combinedData = allResults.flat();
  // 按id去重，确保数据唯一
  this.demandesSaaSEvaluator = [...new Map(combinedData.map(item => [item.id, item])).values()];
});

getEvaluatorSaaSRequests方法无需修改 #

getEvaluatorSaaSRequests(nomBu) {
  return this.firestore
    .collection('solutions', ref => 
      ref.where('businessUnitOfSolution', '==', nomBu).where('type', '==', 'SaaS')
    )
    .snapshotChanges();
}

HTML代码保持原样 #

<tbody>
    <tr *ngFor="let demande of demandesSaaSEvaluator; let i = index">
      <td>{{ demande.id }}</td>
      <td>{{ demande.solutionName }}</td>
      <td>{{ demande.nomBu }}</td>
      <td>{{ demande.user }}</td>
    </tr>
</tbody>

关键细节说明 #

• from：将Firestore的异步Promise转换为Observable，统一用RxJS操作符处理数据流。
• forkJoin：等待所有并行请求完成后返回结果数组，适合一次性获取多组独立数据的场景。
• take(1)：限制Observable仅发射一次数据，避免snapshotChanges的持续监听导致重复添加。
• flat()：将多个请求返回的数组合并为一维数组，适配HTML的*ngFor遍历。
• 去重逻辑：通过Map以数据id为键去重，确保最终展示的数据集无重复项。

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