RxJs：如何在流闲置y秒后每隔x秒发送固定值？

你提到已经用RxJS封装了WebSocket，中间通过Nginx代理（配置60秒闲置断开），而且客户端的重连逻辑运行正常——但这其实意味着你可能要面对几个容易被忽视的隐性问题，比如消息丢失、状态不同步或者频繁重连的资源消耗。我来分享几个针对性的优化思路：

首先明确你可能面临的潜在问题 #

• 传输间隙的消息丢失：如果Nginx触发断开的瞬间，刚好有消息在传输途中，或者重连的间隙里API推送了新消息，客户端会直接错过这些内容
• 客户端与服务端状态偏差：重连后客户端的本地状态和API端的实时状态可能不一致，需要额外的同步逻辑
• 无意义的频繁重连：如果客户端长时间处于闲置状态，会每隔60秒触发一次重连，额外消耗网络和服务端资源

针对性的优化方案 #

1. 主动心跳替代被动重连

与其等Nginx主动断开再触发重连，不如提前发送心跳包维持连接活跃，从根源上减少断开次数：

import { interval, Subject } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

// 用于停止心跳的信号主题
const stopHeartbeat$ = new Subject<void>();

// 每隔55秒发送一次心跳（比Nginx的60秒超时短，避免触发断开）
interval(55000).pipe(takeUntil(stopHeartbeat$)).subscribe(() => {
  // 假设ws$是你的RxJS WebSocket实例
  ws$.next({ type: 'heartbeat', timestamp: Date.now() });
});

// 连接关闭/出错时停止心跳，重连后再重启
ws$.subscribe({
  complete: () => {
    stopHeartbeat$.next();
    // 执行你的重连逻辑
    reconnectWebSocket();
  },
  error: () => {
    stopHeartbeat$.next();
    reconnectWebSocket();
  }
});

2. 待发送消息队列避免丢失

在客户端维护一个消息队列，连接断开时暂存未发送的消息，重连成功后批量补发：

const pendingMessages: any[] = [];

// 封装发送消息的方法
function sendWebSocketMessage(msg: any) {
  if (ws$.closed) {
    pendingMessages.push(msg);
  } else {
    ws$.next(msg);
  }
}

// 重连成功后补发队列中的消息
function reconnectWebSocket() {
  const newWs$ = createWebSocketConnection(); // 你的连接创建方法
  newWs$.subscribe({
    next: (msg) => {
      // 处理收到的消息
      handleIncomingMessage(msg);
    },
    complete: () => reconnectWebSocket(),
    error: () => reconnectWebSocket(),
    // 连接成功后补发消息
    subscribe: () => {
      pendingMessages.forEach(msg => newWs$.next(msg));
      pendingMessages.length = 0; // 清空队列
    }
  });
}

3. 重连后的状态同步机制

重连成功后，主动向API请求当前最新状态，确保客户端与服务端数据一致：

function reconnectWebSocket() {
  const newWs$ = createWebSocketConnection();
  newWs$.pipe(
    // 先发送状态同步请求
    tap(() => newWs$.next({ type: 'request-full-state' })),
    // 过滤出状态同步的响应
    filter(msg => msg.type === 'full-state-response')
  ).subscribe((stateData) => {
    // 更新客户端本地状态
    updateClientLocalState(stateData);
  });
}

额外的Nginx配置建议 #

如果业务允许，可以调整Nginx的超时配置，配合客户端心跳进一步优化：

location /ws-endpoint {
    proxy_pass http://your-api-server;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    # 延长超时时间，比如5分钟，配合客户端55秒一次的心跳
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
}

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