RxJS结合expand与switchMap获取分页学生列表及对应地址的实现问题

核心用到的操作符 #

• concatMap/mergeMap：处理流的嵌套转换，将上游的分页数据、学生对象转换为对应的接口请求流
• from：将每页返回的学生数组转换为逐个发射单个学生对象的Observable
• map：拼接学生基础信息和返回的地址数据，生成完整的学生对象

代码实现 #

原有代码的小优化：expand中结束分页的返回值可以替换为RxJS内置的EMPTY，比手动创建空Observable更规范。
完整代码如下：

// 提前导入依赖
import { EMPTY, from } from 'rxjs';
import { concatMap, expand, map } from 'rxjs/operators';

// 构造带地址的学生数据流
const studentsWithAddress$ = this.searchStudent(nextPageToken).pipe(
  // 原有分页拉取逻辑
  expand((students: StudentsDto) => {
    if (students.nextPageToken) {
      nextPageToken = students.nextPageToken;
      return this.searchStudent(nextPageToken);
    } else {
      return EMPTY;
    }
  }),
  // 把每页的学生数组拆成单个学生逐个发射
  concatMap((page: StudentsDto) => from(page.students)),
  // 给每个学生请求地址，合并到学生对象中
  concatMap((student: StudentDto) => 
    this.getAddress(student.id).pipe(
      map(address => ({
        ...student,
        address
      }))
    )
  )
);

// 订阅获取结果
const subscription = studentsWithAddress$.subscribe({
  next(studentWithAddress) {
    // 此处可直接拿到带地址的学生信息
    console.log(studentWithAddress);
  },
  error(e) {
    console.log(e);
  },
  complete() {
    subscription.unsubscribe();
  }
});

可选优化 #

• 并发控制：如果地址接口没有严格限流要求，可将第二个concatMap替换为mergeMap，还可以指定并发数，示例：mergeMap(student => this.getAddress(student.id).pipe(...), 5)，代表同时最多发起5个地址查询请求，提升拉取速度。
• 全量聚合：如果需要等所有学生的地址都拉取完成后统一处理，可在流的末尾追加toArray()操作符，订阅的next回调会一次性收到完整的带地址的学生数组。

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