RxJS多轮API请求场景下如何仅获取Observable的最后返回值

问题根因 #

1. 原有逻辑固定在首次返回未处理状态时发起10次轮询，哪怕中间某次请求已经返回处理完成也不会停止，且每次请求成功都会调用subject.next，因此所有响应都会下发到组件。
2. 嵌套使用subscribe不符合RxJS响应式编程规范，逻辑不可控，同时存在内存泄漏风险。

解决方案 #

你可以直接使用RxJS原生的expand操作符实现可控的递归轮询，无需手动维护定时器和循环：

第一步：确认请求重试策略配置

确保你的genericRetryStrategy已经配置为请求失败最多重试3次，示例配置如下：

import { Observable, delay, take } from 'rxjs';

export function genericRetryStrategy(maxRetry = 3) {
  return (errors: Observable<any>) => errors.pipe(
    delay(500),
    take(maxRetry)
  );
}

第二步：重写轮询服务逻辑

import { BehaviorSubject, EMPTY, expand, take, delay, filter, finalize } from 'rxjs';

public positionConsolidate() {
    const subject = new BehaviorSubject<any>([]);
    let requestCount = 0;
    // 最大轮询次数（含首次请求）
    const maxRequestCount = 10;

    this.api.getPositionConsolidate()
      .pipe(
        // 递归处理轮询逻辑
        expand((response) => {
          requestCount++;
          // 满足终止条件：处理完成/达到最大请求次数，终止流
          if (response.hasProcessado || requestCount >= maxRequestCount) {
            return EMPTY;
          }
          // 延迟500ms发起下一次请求
          return this.api.getPositionConsolidate().pipe(delay(500));
        }),
        // 过滤中间响应，仅保留最终符合终止条件的结果
        filter(response => response.hasProcessado || requestCount >= maxRequestCount),
        finalize(() => this.loadingService.loadingOff())
      )
      .subscribe({
        next: (finalResponse) => {
          // 仅下发最终结果
          subject.next(finalResponse);
          subject.complete();
        },
        error: (err) => {
          subject.error(err);
          this.loadingService.loadingOff();
        }
      })

    return subject.asObservable();
}

第三步：组件代码无需修改

原有组件内的订阅逻辑可以直接沿用，现在只会收到轮询结束后的最终一次响应。

优化说明 #

• 轮询过程中只要返回hasProcessado == true就会立刻终止请求流，不会发起后续无效请求
• 仅最终满足终止条件的响应会下发到组件，避免多余的响应通知
• 所有逻辑统一在RxJS管道内处理，避免嵌套订阅导致的内存泄漏问题
• 复用原有API层的重试逻辑，满足请求失败最多重试3次的要求

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