如何修复生产环境下前端RXStomp客户端偶尔无法接收消息的问题？

问题背景 #

此前WebSocket运行正常，修改后端WebSocket URL为public/chatsocket后，生产环境出现接收消息不稳定的情况：Socket连接已建立，后端通过simpMessagingTemplate.convertAndSendToUser确认消息已发送，但前端有时能收到、有时收不到，部分用户正常部分异常，测试环境无此问题。

前端配置与代码 #

RXStomp基础配置

import { RxStompConfig } from '@stomp/rx-stomp';
import { environment } from '../../../environments/environment';

export const myRxStompConfig: RxStompConfig = {
  reconnectDelay: 20000,
  debug: (msg: string): void => {
    if (!environment.production) {
      console.log(msg);
    }
  },
};

RxStompService实现

import { Injectable } from "@angular/core";
import { RxStomp } from "@stomp/rx-stomp";
import SockJS from "sockjs-client";
import { environment } from "../../../environments/environment";
import { myRxStompConfig } from "../configurations/rx-stomp.config";

@Injectable({
  providedIn: "root",
})
export class RxStompService extends RxStomp {
  public currentRetry  = 0;

  public resetRetry() {
    this.currentRetry = 0;
  }
}

export function rxStompServiceFactory() {
  const rxStomp = new RxStompService();
  myRxStompConfig.webSocketFactory = function () {
    return new SockJS(`${environment.baseUrl}/public/chatsocket`);
  };
  rxStomp.resetRetry();
  myRxStompConfig.beforeConnect = (): Promise<void> => {
    return new Promise<void>((resolve, reject) => {
      if (rxStomp.currentRetry <= 5) {
        rxStomp.currentRetry++;
        resolve();
      }
    });
  };
  rxStomp.configure(myRxStompConfig);
  rxStomp.activate();
  return rxStomp;
}

前端频道订阅代码

this.rxStompService.watch(`/user/${id}/message`).subscribe((message) => {
   console.log(message.body);
};

后端消息发送代码（Spring Boot） #

private void sendWebMessageToConversation(MessageResponseDto messageResponseDto, String destinationId) {
        try{
            simpMessagingTemplate.convertAndSendToUser(destinationId, "/message", messageResponseDto);
        }
        catch (Exception e){
            logger.error(e.getMessage(), e);
        }
    }

修复步骤 #

1. 修复连接重试逻辑的阻塞问题 #

当前beforeConnect方法中，若currentRetry >5会一直不resolve，导致连接挂起。修改为超过重试次数后明确reject，并添加错误监听重置计数器：

myRxStompConfig.beforeConnect = (): Promise<void> => {
  return new Promise<void>((resolve, reject) => {
    if (rxStomp.currentRetry <= 5) {
      rxStomp.currentRetry++;
      resolve();
    } else {
      reject(new Error("Max reconnection attempts reached"));
    }
  });
};

// 添加连接错误监听，重置重试计数器
rxStomp.stompErrors$.subscribe(() => {
  rxStomp.resetRetry();
});

2. 生产环境开启关键调试日志 #

调整debug配置，生产环境保留订阅、消息接收、连接状态相关日志，方便定位问题：

debug: (msg: string): void => {
  if (!environment.production || msg.includes('SUBSCRIBE') || msg.includes('MESSAGE') || msg.includes('CONNECTED') || msg.includes('ERROR')) {
    console.log(msg);
  }
},

3. 确保用户会话ID一致性 #

Spring Boot的convertAndSendToUser依赖会话ID匹配，需确认：

• 前端WebSocket请求携带正确的会话Cookie（生产环境需检查跨域Cookie配置）
• 后端Stomp端点配置允许会话传递：

@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
  registry.addEndpoint("/public/chatsocket")
          .setAllowedOriginPatterns("*")
          .withSockJS()
          .setSessionCookieNeeded(true);
}

4. 订阅逻辑添加错误处理与重订阅 #

避免订阅异常导致后续消息无法接收，添加错误回调与自动重订阅：

this.rxStompService.watch(`/user/${id}/message`).subscribe({
  next: (message) => {
    console.log(message.body);
  },
  error: (err) => {
    console.error('订阅频道失败:', err);
    setTimeout(() => {
      this.subscribeMessageChannel(id);
    }, 3000);
  }
});

5. 检查生产环境代理与网络配置 #

确认反向代理（如Nginx）的WebSocket支持配置，避免连接被提前断开：

location /public/chatsocket {
    proxy_pass http://your-backend-service;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
    proxy_connect_timeout 60s;
    proxy_send_timeout 60s;
    proxy_read_timeout 600s;
}

同时检查防火墙、负载均衡的超时设置，确保长连接不被强制断开。

6. 监听连接状态自动重订阅 #

添加连接状态监听，在连接重建后自动重新订阅频道：

this.rxStompService.connected$.subscribe((isConnected) => {
  if (isConnected) {
    this.subscribeMessageChannel(id);
  }
});

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