Angular中合并多个Observable订阅结果及解决重复数据问题

问题背景 #

业务流程：

1. 从businessUnit集合查询evaluators数组包含当前用户邮箱的文档，得到多个nomBu字符串，此步骤正常。
2. 利用这些nomBu调用getEvaluatorSaaSRequests方法，从solutions集合获取对应SaaS请求元数据，此步骤正常。
3. 需在HTML仪表盘展示所有数据，但遇到两个问题：
   • 原代码仅显示最后一个Observable的执行结果。
   • 改用push合并结果后，数据全展示但出现重复。

原代码问题分析 #

原组件代码（步骤1-3） #

demandesSaaSEvaluator: Demande[];

this.firestore.collection('businessUnit')
  .where('evaluators', 'array-contains', this.auth.currentUserEmail).get()
  .then(snapshot => {
    snapshot.forEach(doc => {
      this.service.getEvaluatorSaaSRequests(doc.data().nomBu).subscribe(actionArray => {
        this.demandesSaaSEvaluator = actionArray.map(item => {
          return {
            id: item.payload.doc.id,
            ...(item.payload.doc.data() as object)
          } as Demande;
        });
      });
    });
  });

• 问题：每次订阅Observable时直接赋值this.demandesSaaSEvaluator = ...，后续订阅会覆盖之前的数组，最终只保留最后一组数据。

临时push方案代码 #

demandesSaaSEvaluator: Demande[] = [];

this.service.getEvaluatorSaaSRequests(doc.data().nomBu).subscribe(actionArray => {               
  this.demandesSaaSEvaluator.push(...actionArray.map(item => {
    return {
      id: item.payload.doc.id,
      ...(item.payload.doc.data() as object)
    } as Demande;      
  }));
});

• 问题：snapshotChanges()是实时监听，当solutions集合中对应数据更新时，会再次触发subscribe，新数据被重复push；若存在重复的nomBu，也会导致重复数据。

服务层方法代码 #

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>

解决方案 #

方案1：仅一次性获取数据（无实时更新） #

适合不需要实时同步数据的场景，改用get()单次获取数据，结合forkJoin合并所有请求结果，避免覆盖和重复。

修改服务层方法

getEvaluatorSaaSRequests(nomBu) {
  return this.firestore
    .collection('solutions', ref => 
      ref.where('businessUnitOfSolution', '==', nomBu).where('type', '==', 'SaaS')
    )
    .get() // 改用get()获取单次数据，而非实时监听
    .pipe(
      map(snapshot => {
        return snapshot.docs.map(doc => {
          return {
            id: doc.id,
            ...doc.data() as object
          } as Demande;
        });
      })
    );
}

组件层代码

import { forkJoin, Subscription } from 'rxjs';

// ...

demandesSaaSEvaluator: Demande[] = [];
private requestSub: Subscription = new Subscription();

// ...

this.firestore.collection('businessUnit')
  .where('evaluators', 'array-contains', this.auth.currentUserEmail)
  .get()
  .then(snapshot => {
    // 收集所有nomBu对应的请求Observable
    const requestObservables = snapshot.docs.map(doc => {
      const nomBu = doc.data().nomBu;
      return this.service.getEvaluatorSaaSRequests(nomBu);
    });

    // 合并所有请求，等待全部完成后合并结果
    this.requestSub.add(
      forkJoin(requestObservables).subscribe(allResults => {
        // 把多维数组扁平化为一维数组
        this.demandesSaaSEvaluator = allResults.flat();
      })
    );
  });

// 组件销毁时清理订阅，避免内存泄漏
ngOnDestroy() {
  this.requestSub.unsubscribe();
}

方案2：实时监听数据（自动同步更新） #

需要实时同步solutions数据变化的场景，用merge合并所有实时Observable，同时通过ID去重避免重复数据。

组件层代码

import { merge, Subscription } from 'rxjs';
import { map } from 'rxjs/operators';

// ...

demandesSaaSEvaluator: Demande[] = [];
private realtimeSub: Subscription = new Subscription();

// ...

this.firestore.collection('businessUnit')
  .where('evaluators', 'array-contains', this.auth.currentUserEmail)
  .get()
  .then(snapshot => {
    // 收集所有nomBu对应的实时Observable，并提前映射为Demande数组
    const realtimeObservables = snapshot.docs.map(doc => {
      const nomBu = doc.data().nomBu;
      return this.service.getEvaluatorSaaSRequests(nomBu).pipe(
        map(actionArray => {
          return actionArray.map(item => {
            return {
              id: item.payload.doc.id,
              ...item.payload.doc.data() as object
            } as Demande;
          });
        })
      );
    });

    // 合并所有实时流，任意一个流更新时触发
    this.realtimeSub.add(
      merge(...realtimeObservables).subscribe(newResults => {
        // 去重逻辑：先移除已有相同ID的数据，再添加新数据（支持更新）
        this.demandesSaaSEvaluator = [
          ...this.demandesSaaSEvaluator.filter(d => !newResults.some(n => n.id === d.id)),
          ...newResults
        ];
      })
    );
  });

// 组件销毁时清理订阅
ngOnDestroy() {
  this.realtimeSub.unsubscribe();
}

注意事项 #

• 始终清理RxJS订阅，避免内存泄漏。
• 若businessUnit集合中的用户关联nomBu可能变化，可考虑对businessUnit也添加实时监听，动态更新请求流。

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