RxJS中Socket重连重试次数限制及计数重置方案问询

你遇到的这个问题很典型——当前用take(this.retryCount)是在整个错误流的生命周期内限制总重试次数，而不是每次连接断开后独立计数。所以第一次断开重试4次成功后，计数已经用掉了4次，下次断开就只剩下1次重试机会了。

要实现“每次断开后都拥有独立的5次重试机会”，核心是在连接成功时重置重试计数，让下次断开时重新从0开始计数。这里有个清晰的实现方案：

// 创建一个用于重置计数的Subject
private resetRetryCount$ = new Subject<void>();

// Socket连接逻辑
socket.pipe(
  tap((data) => {
    console.log(data);
    // 每次成功接收数据（说明连接正常），发送重置信号
    this.resetRetryCount$.next();
  }),
  retryWhen((errors) =>
    errors.pipe(
      // 用scan跟踪当前断开后的重试次数
      scan((currentCount) => {
        if (currentCount >= this.retryCount) {
          // 达到重试上限，抛出错误终止流
          throw new Error(`已达到最大重试次数：${this.retryCount}`);
        }
        return currentCount + 1;
      }, 0),
      // 按重试次数递增延迟时间（指数退避）
      delayWhen((count) => timer(count * 1000)),
      // 收到重置信号时，终止当前重试流，下次断开重新计数
      takeUntil(this.resetRetryCount$)
    )
  )
).subscribe({
  error: (err) => console.error("重连失败并终止：", err)
});

为什么这个方案有效？ #

1. 重置信号触发：每次连接成功并收到数据时，resetRetryCount$.next()会触发takeUntil终止当前的重试流。当下一次连接断开时，retryWhen会重新订阅错误流，scan会从0开始重新计数。
2. 独立计数逻辑：scan的计数仅针对当前断开后的重试次数，不会和之前的断开重试累计。
3. 上限可控：当当前断开后的重试次数达到设定值时，直接抛出错误终止流，你也可以根据需求调整这里的逻辑（比如改为进入离线模式而非终止）。

另一种无额外Subject的简洁写法 #

如果你不想引入额外的Subject，也可以通过嵌套流的方式让每次断开后的重试逻辑独立：

socket.pipe(
  tap((data) => console.log(data)),
  retryWhen((errors) =>
    errors.pipe(
      // 每次错误都创建一个全新的重试流，最多重试retryCount次
      switchMap(() =>
        timer(0, 1000).pipe(
          take(this.retryCount),
          delayWhen((count) => timer(count * 1000))
        )
      )
    )
  )
).subscribe();

这个方案的原理是：每次连接断开触发错误时，switchMap会创建一个最多发出retryCount个值的新流。如果中途连接成功，这个新流就会被取消，下次断开时再创建全新的流，自然实现了计数重置。

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