如何监听WebSocketSubject的重连事件？连接异常问题求助

我之前也碰到过一模一样的问题！WebSocketSubject在网络突然断开的场景下，确实不会立刻触发closeObserver或者openObserver，连retryWhen里的日志都没输出——这其实是浏览器底层WebSocket API的特性导致的：当网络突然中断时，浏览器不会立刻抛出错误或触发onclose事件，而是会默默尝试重连几次（或者等待一段超时时间），这段时间里你的RxJS钩子完全感知不到状态变化。

问题根源 #

浏览器原生WebSocket在网络断开后，不会主动通知上层应用连接已断，而是会等待TCP层的超时（通常是几十秒甚至更久），这就导致你的closeObserver和retryWhen都无法及时捕获断开事件。

靠谱的解决方案：手动实现心跳机制 #

要解决这个问题，最可靠的方式是添加心跳检测——定时给服务器发送心跳包，监听服务器的响应，如果超过指定时间没收到响应，就主动关闭WebSocket连接，这样就能触发closeObserver和retryWhen的重连逻辑了。

下面是完整的可运行代码示例：

import { WebSocketSubject, WebSocketSubjectConfig, retryWhen, delay, tap, interval, takeUntil } from 'rxjs';

// 配置WebSocketSubject的基础参数
const wsConfig: WebSocketSubjectConfig<any> = {
  url: 'wss://your-websocket-server-url',
  // 连接打开时的钩子
  openObserver: {
    next: (event) => {
      console.log('✅ WebSocket连接已建立');
      // 连接成功后立即启动心跳检测
      startHeartbeat();
    }
  },
  // 连接关闭时的钩子
  closeObserver: {
    next: (event) => {
      console.log('❌ WebSocket连接已关闭，准备重连');
      // 关闭心跳定时器
      stopHeartbeat();
    }
  },
  // 错误捕获钩子
  errorObserver: {
    next: (error) => {
      console.error('⚠️ WebSocket发生错误:', error);
    }
  }
};

// 创建WebSocketSubject实例
const wsSubject = new WebSocketSubject(wsConfig);

// 心跳相关配置
let heartbeatTimer: any;
const HEARTBEAT_SEND_INTERVAL = 5000; // 每5秒发一次心跳
const HEARTBEAT_TIMEOUT = 10000; // 10秒没收到响应则判定断开

// 启动心跳检测
function startHeartbeat() {
  // 先清除可能存在的旧定时器
  stopHeartbeat();

  heartbeatTimer = setInterval(() => {
    // 发送心跳包（格式和服务器约定好）
    wsSubject.next({ type: 'heartbeat' });

    // 设置心跳超时定时器，超时则主动断开连接
    const timeoutTimer = setTimeout(() => {
      console.log('⏰ 心跳超时，主动断开连接');
      wsSubject.complete(); // 主动关闭连接，触发closeObserver
    }, HEARTBEAT_TIMEOUT);

    // 监听服务器的心跳响应，收到则清除超时定时器
    wsSubject.pipe(takeUntil(interval(HEARTBEAT_SEND_INTERVAL))).subscribe({
      next: (msg) => {
        if (msg.type === 'heartbeat-response') {
          clearTimeout(timeoutTimer);
        }
      }
    });
  }, HEARTBEAT_SEND_INTERVAL);
}

// 停止心跳检测
function stopHeartbeat() {
  if (heartbeatTimer) {
    clearInterval(heartbeatTimer);
    heartbeatTimer = null;
  }
}

// 配置重连逻辑
wsSubject.pipe(
  retryWhen((errors) => 
    errors.pipe(
      tap((err) => console.error('🔄 准备重连，错误原因:', err)),
      delay(3000) // 间隔3秒后尝试重连
    )
  )
).subscribe({
  next: (msg) => console.log('📥 收到消息:', msg),
  error: (err) => console.error('❌ 订阅错误:', err),
  complete: () => console.log('🔚 订阅完成')
});

关键细节说明 #

1. 心跳机制的必要性：通过主动发送心跳并监听响应，我们能主动感知到连接是否存活，避免等待浏览器的超时
2. 主动触发重连：当心跳超时后，调用wsSubject.complete()主动关闭连接，这样就能触发closeObserver，进而让retryWhen捕获到错误并执行重连
3. 服务器配合：需要服务器端实现心跳响应逻辑，收到心跳包后返回对应的响应消息（比如{ type: 'heartbeat-response' }）

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