带退避重试的RxJS轮询自定义算子实现问题求助

问题分析 #

你当前的实现存在两个核心问题：

1. 轮询与重试逻辑耦合：使用timer(0, delay)定时触发请求，会导致请求在队列中排队，即使前一个请求（含重试）尚未完成，不符合"请求耗时较长时不触发其他请求"的要求。
2. 重试时异常输出：原代码中retryWhen的错误流未正确处理，可能导致错误对象被意外暴露，引发控制台异常打印。

同时，原实现可以进一步简化，将轮询与重试逻辑解耦，让代码更易读和维护。

修正后的实现 #

以下是满足所有需求的简化版算子，逻辑清晰且完全符合你的要求：

import { Observable, retryWhen, scan, map, delayWhen, timer, repeatWhen, delay } from 'rxjs';

function repeatWithBackoff<T>(delay: number, maxDelay = 60000) {
  return (source: Observable<T>) =>
    source.pipe(
      // 处理失败重试：指数退避，直到达到最大延迟
      retryWhen((errors) =>
        errors.pipe(
          // 统计重试次数，从0开始
          scan((attemptCount) => attemptCount + 1, 0),
          // 计算退避延迟：首次delay，之后翻倍，不超过maxDelay
          map((count) => Math.min(delay * Math.pow(2, count), maxDelay)),
          // 等待对应延迟后重试
          delayWhen((backoffDelay) => timer(backoffDelay))
        )
      ),
      // 处理轮询：请求成功完成后，等待delay毫秒再发起下一次
      repeatWhen((completed$) => completed$.pipe(delay(delay)))
    );
}

逻辑说明 #

这个实现完全覆盖你的四个需求：

• 轮询机制：通过repeatWhen实现，每次请求（含重试成功）完成后，等待delay毫秒再发起下一次请求，确保同一时间只有一个请求在执行。
• 指数退避重试：
  • 首次失败等待delay毫秒重试
  • 再次失败延迟翻倍，直到达到maxDelay，后续重试均使用该最大延迟
• 重试计数器重置：每次重试成功后，retryWhen的流会终止，下一次失败时scan的计数器会从0重新开始，自动重置延迟计算。
• 串行请求：retryWhen和repeatWhen的组合确保所有请求串行执行，请求耗时较长时不会触发新的请求。

使用方式 #

和你原来的用法完全一致：

httpClient.post('your-api-url', payload)
  .pipe(repeatWithBackoff(1000, 60000))
  .subscribe(
    (result) => console.log('请求结果:', result),
    (error) => console.error('终止错误（如非重试类错误）:', error)
  );

可选优化 #

如果需要针对特定错误停止重试（比如404、500等），可以在retryWhen中添加filter操作符：

retryWhen((errors) =>
  errors.pipe(
    // 仅对特定错误进行重试，比如网络错误或5xx状态码
    filter((err) => err.status >= 500 || err.status === 0),
    scan((attemptCount) => attemptCount + 1, 0),
    map((count) => Math.min(delay * Math.pow(2, count), maxDelay)),
    delayWhen((backoffDelay) => timer(backoffDelay))
  )
)

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