如何实现Observable同步执行？嵌套订阅执行顺序异常排查

问题表现 #

代码执行时console.log('2')触发时机早于内层请求回调的console.log('1')，二级分类数据集catalogueCategorieLayer2打印时未完成异步请求赋值，不符合顺序执行预期。

根因分析 #

• Observable的.subscribe()是非阻塞异步调用：发起二级分类GQL请求后，不会等待请求返回，会直接执行后续同步代码（也就是console.log('2')），等请求响应到达后才会进入subscribe回调执行赋值和console.log('1')，顺序天然不符合同步代码的执行预期。
• 多层嵌套手写subscribe是RxJS典型反模式，会导致回调地狱、执行时序不可控、订阅无法统一取消引发内存泄漏，也没法统一处理加载状态和错误。
• 现有代码中三级分类请求和一、二级请求完全独立发起，没有依赖约束，也会出现三级数据早于一、二级数据加载完成的时序混乱问题。

原问题代码 #

服务层代码 #

getCatalogueCategoriesByLayerAndParentForPortal(layer: number, target: TargetAciInput): Observable<CatalogueCategoryType[]> {
  return this.getCatalogueCategoriesByLayerAndParentForPortalGQL
    .fetch({ layer, target })
    .pipe(map((response: any) => response.data.getCatalogueCategoriesByLayerAndParentForPortal));
}

getCatalogueCategoriesByLayerAndParentForPortals(
  layer: number,
  target: TargetAciInput,
  parent: string,
): Observable<CatalogueCategoryType[]> {
  return this.getCatalogueCategoriesByLayerAndParentForPortalGQL
    .fetch({ layer, target, parent })
    .pipe(map((response: any) => response.data.getCatalogueCategoriesByLayerAndParentForPortal));
}

NgOnInit原实现代码 #

this.appService.getCatalogueCategoriesByLayerAndParentForPortal(1, { pos: '' }).subscribe((res: any) => {
  res.forEach((element) => {
    this.catalogueCategorieLayer1.push(element);
    this.appService.getCatalogueCategoriesByLayerAndParentForPortals(2, { pos: '' }, this.parentId).subscribe((response: any) => {
      this.catalogueCategorieLayer2 = [...this.catalogueCategorieLayer2, response];
      console.log('1');
    });
    console.log('2', this.catalogueCategorieLayer2);
    this.hasChildren = element.hasChildren;
  });
});

this.appService.getCatalogueCategoriesByLayerAndParentForPortals(3, { pos: '' }, this.parentId2).subscribe((response: any) => {
  response.forEach((element) => {
    this.catalogueCategorieLayer3[0] = element.name;
  });
});

修复方案 #

用RxJS高阶操作符替换嵌套subscribe，按数据依赖关系串联请求流程，保证执行顺序符合预期：

• 用tap操作符处理请求返回后的同步赋值逻辑，不改变流的数据
• 用mergeMap做流的切换，等上一级请求完成后再发起下一级依赖请求
• 用forkJoin合并同一层级的多个并行请求，等所有同层级请求全部返回后再做后续处理
• 统一在最外层订阅处理最终数据、错误逻辑，避免嵌套订阅

修复后NgOnInit代码 #

import { forkJoin } from 'rxjs';
import { mergeMap, tap } from 'rxjs/operators';

// 按依赖顺序串联请求
this.appService.getCatalogueCategoriesByLayerAndParentForPortal(1, { pos: '' })
  .pipe(
    // 一级分类返回后先赋值一级数据
    tap((layer1List: CatalogueCategoryType[]) => {
      this.catalogueCategorieLayer1 = layer1List;
      this.hasChildren = layer1List.some(item => item.hasChildren);
    }),
    // 一级数据处理完，批量发起所有一级分类对应的二级分类请求
    mergeMap((layer1List: CatalogueCategoryType[]) => {
      // 生成所有二级分类请求的Observable集合
      const layer2Requests = layer1List.map(item => {
        // 注意：原代码传的this.parentId如果是当前一级项的ID，替换为item.id即可
        return this.appService.getCatalogueCategoriesByLayerAndParentForPortals(2, { pos: '' }, item.id);
      });
      // 等待所有二级请求全部返回后再进入下一步
      return forkJoin(layer2Requests);
    }),
    // 所有二级请求完成后赋值二级数据
    tap((layer2Result: CatalogueCategoryType[][]) => {
      // 扁平化所有二级分类结果
      this.catalogueCategorieLayer2 = layer2Result.flat();
      console.log('1');
      console.log('2', this.catalogueCategorieLayer2);
    }),
    // 二级数据处理完，发起三级分类请求
    mergeMap(() => {
      // 如果parentId2是从二级数据中取的动态值，直接在这一步从上层作用域获取即可
      return this.appService.getCatalogueCategoriesByLayerAndParentForPortals(3, { pos: '' }, this.parentId2);
    })
  )
  .subscribe({
    next: (layer3List: CatalogueCategoryType[]) => {
      // 赋值三级分类数据
      this.catalogueCategorieLayer3 = layer3List.map(item => item.name);
    },
    error: (err) => {
      // 统一处理全链路请求错误
      console.error('分类数据加载失败', err);
    }
  });

修复要点说明 #

• 所有代码严格按照请求依赖顺序执行，console.log('2')会在二级数据全部赋值完成后触发，完全符合预期顺序
• 消除了嵌套订阅的问题，后续加全局loading、请求取消、错误拦截都更方便，也避免了内存泄漏风险
• 不要在异步流程里依赖全局this上的动态变量传参，所有参数直接从上一级请求的返回值里取，避免时序问题拿到旧值
• 如果三级分类请求不依赖一、二级数据，可以在最外层用forkJoin把一级流程和三级流程合并为并行请求，提升加载速度

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