Spring WebSocket(STOMP+RabbitMQ)批量消息发送性能异常及断连问题求助

• 同步批量发送引发消息积压：当前代码用同步for循环瞬间推送1000条消息，浏览器WebSocket发送缓冲区容量有限，后续发送请求会被阻塞排队。重复发送时，前一轮未处理完的消息会叠加新消息，导致队列积压越来越严重，耗时逐次攀升。
• UI线程阻塞与内存泄漏：批量消息接收后若同步更新DOM，会阻塞浏览器主线程，引发界面卡顿、闪烁。此外，若重连时未清理旧的STOMP监听、无用DOM元素，会造成内存泄漏，随着重复次数增加，内存占用飙升，GC频繁触发，进一步加剧性能问题，甚至导致WebSocket会话因资源耗尽断开。
• 消息代理与后端处理瓶颈：短时间大量消息涌入RabbitMQ，若后端消费速度跟不上，会造成队列积压，RabbitMQ会触发流量控制机制限制客户端发送速率，导致发送变慢。重连后，队列中留存的未处理旧消息叠加新消息，压力进一步增大，耗时更长。
• 缺失流量控制机制：客户端未基于后端处理能力实现流量控制，盲目持续发送，导致消息堆积超出系统承载上限，最终触发会话超时断开。

1. 异步分批发送消息
   替换同步循环为异步分批发送，给浏览器和后端留出处理时间，示例代码：

   async function sendBatchMessages(totalCount, batchSize = 10, delay = 100) {
     if (!stompClient) return;
     for (let i = 1; i <= totalCount; i++) {
       const chatMessage = {
         sender: name,
         content: `File Processing at ${i} %`,
         type: 'Process'
       };
       stompClient.send("/app/chat.sendMessage", {}, JSON.stringify(chatMessage));
       // 每发送指定批次后等待固定时长
       if (i % batchSize === 0) {
         await new Promise(resolve => setTimeout(resolve, delay));
       }
     }
   }
   // 调用示例
   sendBatchMessages(999);
   

2. 优化UI处理逻辑

   • 用requestAnimationFrame批量更新DOM，减少重绘重排：

     let messageBuffer = [];
     function handleMessageReceived(message) {
       const parsedMsg = JSON.parse(message.body);
       messageBuffer.push(parsedMsg);
       requestAnimationFrame(() => {
         if (messageBuffer.length === 0) return;
         const messagesContainer = document.getElementById('messages');
         const fragment = document.createDocumentFragment();
         messageBuffer.forEach(msg => {
           const div = document.createElement('div');
           div.textContent = `${msg.sender}: ${msg.content}`;
           fragment.appendChild(div);
         });
         messagesContainer.appendChild(fragment);
         messageBuffer = [];
       });
     }
     

   • 重连前清理旧资源，避免内存泄漏：

     function disconnect() {
       if (stompClient) {
         stompClient.unsubscribe();
         stompClient.disconnect();
         stompClient = null;
       }
       document.getElementById('messages').innerHTML = '';
       messageBuffer = [];
     }
     

3. RabbitMQ与后端优化

   • 配置RabbitMQ队列：设置x-max-length限制队列最大长度，添加x-message-ttl自动清理超时未处理消息，避免无限积压。
   • 提升后端消费能力：增加消费者实例，优化消息处理逻辑，确保后端及时发送ACK，避免消息重复堆积。

4. 实现STOMP背压机制
   基于STOMP消息确认机制，收到后端ACK后再发送下一批消息（需后端支持ack模式）：

   let isSending = false;
   async function sendWithBackpressure(totalCount, batchSize = 10) {
     if (!stompClient || isSending) return;
     isSending = true;
     let current = 1;
     while (current <= totalCount) {
       const batchPromises = [];
       for (let i = 0; i < batchSize && current <= totalCount; i++) {
         const msg = {
           sender: name,
           content: `File Processing at ${current} %`,
           type: 'Process'
         };
         batchPromises.push(new Promise(resolve => {
           stompClient.send("/app/chat.sendMessage", {}, JSON.stringify(msg), (err) => {
             err ? console.error(err) : resolve();
           });
         }));
         current++;
       }
       await Promise.all(batchPromises);
     }
     isSending = false;
   }
   

5. 优化重连逻辑
   重连后等待连接稳定再发送消息，避免加重系统负担：

   let isReconnectReady = false;
   function reconnect() {
     setTimeout(() => {
       const socket = new SockJS('/ws');
       stompClient = Stomp.over(socket);
       stompClient.connect({}, (frame) => {
         stompClient.subscribe('/topic/public', handleMessageReceived);
         // 等待1秒确保连接稳定
         setTimeout(() => isReconnectReady = true, 1000);
       }, (error) => {
         console.error('Reconnect failed:', error);
         reconnect();
       });
     }, 3000);
   }
   

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