RxJS 顺序处理HTTP Post请求并实现失败自动重试的问题咨询

问题原因 #

原有代码将retryWhen放在整个流的顶层pipe中，一旦某个请求报错会触发整个subject流重试，无法实现仅重试当前失败的单个请求的需求。

修正后实现方案 #

将重试逻辑封装在concatMap内部，每个请求独立走重试流程，成功后才会处理下一个请求：

import { from, concatMap, retryWhen, tap, delayWhen, timer } from 'rxjs';

// 核心流逻辑
subject
  .pipe(
    concatMap(request => 
      // 把请求Promise转为Observable，单个请求独立配置重试规则
      from(this._sendPostRequest(request)).pipe(
        retryWhen(errors =>
          errors.pipe(
            tap(error => this._logger.error('请求发送失败，5秒后重试', error)),
            delayWhen(() => timer(5000))
          )
        )
      )
    )
  )
  .subscribe();

实现说明 #

• concatMap天然保证请求的串行执行顺序：只有前一个内部Observable（也就是当前请求）完成后，才会拉取subject队列中的下一个请求处理
• 重试逻辑仅作用于当前单个请求，不会影响整个上层流的运行，也不会打乱队列请求的处理顺序
• 未配置重试次数上限，会一直重试直到当前请求成功，符合需求

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