RxJS如何实现等待checkStatus响应后再触发下一次调用

核心问题分析 #

你原有代码的核心问题有两个：

• checkStatus是异步请求方法，你调用后直接判断jobStatus，此时请求还未返回，判断的是旧值，逻辑完全失效
• 用interval实现的轮询不管上一次请求是否完成，每5秒就发起新请求，容易出现请求堆积，也达不到「等上一次请求返回再发起下一次」的要求

改造步骤 #

第一步：改造checkStatus方法，返回Observable供外层控制 #

不要在checkStatus内部订阅，改成返回请求的Observable，把状态赋值逻辑放到tap操作符里：

checkStatus(taskRequestId: number) {
  this.checkCount++;
  return this.vendorWebApiService
    .getRecognition(taskRequestId, this.executiveChangeId)
    .pipe(
      take(1),
      tap((recognitionResponse) => {
        this.jobStatus = recognitionResponse.jobStatus;
        if (recognitionResponse.jobStatus === "Completed") {
          this.recognitionData = recognitionResponse;
        }
      })
    );
}

第二步：改用RxJS expand操作符重写轮询逻辑 #

expand操作符天生支持「等上一个流完成再调用下一个流」的递归逻辑，完美匹配你的需求：

import { EMPTY } from 'rxjs';
import { expand, delay } from 'rxjs/operators';

getRecognitionById() {
  this.loaderService.show(null, true);
  this.checkCount = 0; // 每次发起新任务时重置计数
  this.vendorWebApiService
    .createRecognition(this.executiveChangeId)
    .pipe(take(1))
    .subscribe((res) => {
      // 发起第一次状态检查
      this.checkStatus(res.taskRequestId).pipe(
        // 递归发起下一次检查
        expand(() => {
          // 终止条件1：任务完成
          if (this.jobStatus === "Completed") {
            this.getLatestFeedback();
            this.loaderService.hide(true);
            return EMPTY;
          }
          // 终止条件2：任务失败
          if (this.jobStatus === "Failed") {
            this.loaderService.hide(true);
            alert("Recognition failed. Try again later");
            return EMPTY;
          }
          // 终止条件3：超过最大重试次数36次
          if (this.checkCount >= 36) {
            this.loaderService.hide(true);
            alert("Recognition failed. Try again later");
            return EMPTY;
          }
          // 未满足终止条件：等5秒后发起下一次检查
          return this.checkStatus(res.taskRequestId).pipe(delay(5000));
        })
      ).subscribe();
    });
}

改造说明 #

• 所有轮询逻辑都保证等上一次checkStatus请求返回后，才会判断是否要发起下一次请求，完全符合你的要求
• 不再需要手动管理refresher$和refreshSub订阅，流终止时会自动释放资源
• 所有终止逻辑收敛到一处，避免重复判断，可维护性更高

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