Angular递归Observable数组：complete提前触发等问题排查

核心错误与概念误解 #

1. forkJoin 不执行请求的原因 #

forkJoin 的生效条件是所有输入 Observable 都至少发出一个值并完成。你的代码中，当某个 Site 既不需要更新/创建，也没有子节点时，upsertSites 会返回 combineLatest(site.children.map(...))，而空数组的 combineLatest 会直接完成但不发出任何值。这种无值的 Observable 会导致 forkJoin 无法收集到所有输入的结果，最终直接触发 complete 却不执行任何请求。

2. combineLatest 提前触发 complete 的问题 #

combineLatest 的 complete 事件只会在所有输入 Observable 都完成时触发，但你的代码存在两个关键问题：

• 无操作无子女的 Site 返回的 combineLatest([]) 会提前完成，但此时其他请求仍在进行，combineLatest 不会触发 complete——你可能把 next 回调的触发当成了 complete。
• upsertSites 中 mergeMap 的使用会让父节点请求完成后立即订阅子节点的处理流，导致整个 Observable 发出多次值，而非一次性完成，这容易让你误以为流程提前结束。

3. expand 导致循环重复创建的原因 #

expand 是递归地将每个发出的值映射为新的 Observable，若没有明确的终止条件（比如处理到无子女的节点时返回 EMPTY），或没有标记已处理的节点，就会陷入无限递归，重复创建相同实例。

代码修正方案 #

修正 upsertSites 函数 #

import { Observable, of, forkJoin, catchError, concatMap } from 'rxjs';

upsertSites(site: Site, parentId?: number): Observable<any[]> {
  const sourceSite: Site | undefined = this.sourceSites
    .map(s => s.findSite(site.id).at(-1))
    .filter(s => !!s)
    .at(-1);

  // 情况1：需要更新现有节点
  if (sourceSite && !this.isSiteDeepEqualWithoutChildren(sourceSite, site)) {
    return this.siteService.updateSite(site.id, site.label, parentId).pipe(
      catchError(error => {
        console.error('更新节点失败:', error);
        // 业务需求：更新失败时可选择抛出错误或返回空数组终止当前分支
        return of([]);
      }),
      // 用 concatMap 确保父节点更新完成后再处理子节点
      concatMap(() => {
        if (site.children.length === 0) {
          return of([]);
        }
        // 用 forkJoin 并行处理子节点，确保所有子节点完成后才返回
        return forkJoin(site.children.map(child => this.upsertSites(child, site.id)));
      })
    );
  }
  // 情况2：需要创建新节点
  else if (!sourceSite) {
    return this.siteService.createSite(site.label, parentId).pipe(
      concatMap(data => {
        site.id = data.id;
        if (site.children.length === 0) {
          return of([data]);
        }
        return forkJoin(site.children.map(child => this.upsertSites(child, site.id)));
      })
    );
  }
  // 情况3：无需更新/创建，直接处理子节点
  else {
    if (site.children.length === 0) {
      // 返回 of([]) 确保 Observable 发出一个值后完成，适配 forkJoin 要求
      return of([]);
    }
    return forkJoin(site.children.map(child => this.upsertSites(child, site.id)));
  }
}

修正 save 函数 #

改用 forkJoin，确保所有节点处理完成后才触发 complete：

function save(): void {
  const upsertObservables = this.updatedSites.map(s => this.upsertSites(s));
  forkJoin(upsertObservables).subscribe({
    next: (results) => {
      console.log('所有节点保存完成:', results);
    },
    complete: () => {
      // 此处逻辑会在所有请求+子节点处理完成后触发
      // 执行你的后续操作
    },
    error: (err) => {
      console.error('保存流程出错:', err);
    }
  });
}

expand 的正确用法示例（可选） #

如果想用 expand 处理递归结构，必须设置终止条件：

import { EMPTY } from 'rxjs';

function processSite(site: Site, parentId?: number): Observable<any> {
  // 先处理当前节点
  const current$ = sourceSite 
    ? this.siteService.updateSite(site.id, site.label, parentId) 
    : this.siteService.createSite(site.label, parentId);
  
  return current$.pipe(
    expand((data) => {
      const children = site.children;
      if (children.length === 0) {
        // 无子女，终止递归
        return EMPTY;
      }
      // 并行处理所有子节点，返回合并后的 Observable
      return forkJoin(children.map(child => processSite(child, data.id)));
    })
  );
}

关键概念纠正 #

• forkJoin：适合并行执行多个独立的 Observable，等待所有都完成后返回合并结果，要求每个 Observable 至少发出一个值。
• combineLatest：只要有一个 Observable 发出值就会触发，适合实时合并多个流的最新值的场景，而非等待所有流完成。
• expand：递归处理流中的值，必须明确终止条件（返回 EMPTY），否则会无限递归。

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