别再让用户等消息了！用Spring Boot + WebSocket + RabbitMQ手把手搭建一个高并发消息推送服务

高并发实时消息推送系统实战Spring Boot整合WebSocket与RabbitMQ在电商秒杀、直播互动、在线客服等场景中实时消息推送已成为提升用户体验的关键技术。传统轮询方式不仅浪费服务器资源在高并发场景下更容易导致系统崩溃。本文将带你从零构建一个基于Spring Boot的分布式消息推送系统通过WebSocket实现双向通信结合RabbitMQ的队列特性完成流量削峰最终打造出可承受百万级并发的生产级解决方案。1. 技术选型与架构设计实时消息系统面临三个核心挑战连接稳定性、消息可靠性和系统扩展性。我们采用分层架构设计各组件分工明确通信层WebSocket协议实现全双工通信相比HTTP长轮询节省80%以上的网络开销缓冲层RabbitMQ队列作为消息缓冲区突发流量时保护后端系统存储层Redis集群存储离线消息MySQL持久化重要业务数据技术栈对比表格技术选项适用场景吞吐量延迟学习成本WebSocket实时双向通信10万/s100ms低RabbitMQ异步消息队列5万/s10-100ms中Kafka日志流处理百万/s10-1000ms高Redis Pub/Sub简单消息广播10万/s1ms低提示选择RabbitMQ而非Kafka的原因在于其内置的队列管理功能和更友好的Java生态支持对于需要严格消息顺序的场景更为适合。2. 核心实现步骤2.1 环境准备与基础配置首先在Spring Boot项目中添加必要依赖!-- WebSocket支持 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency !-- RabbitMQ集成 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency !-- Redis缓存 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency配置WebSocket端点注意添加心跳检测配置Configuration EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer { Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(messageHandler(), /push) .setAllowedOrigins(*) .addInterceptors(new AuthInterceptor()) .setHandshakeHandler(new DefaultHandshakeHandler()) .withSockJS() .setHeartbeatTime(30000); // 30秒心跳 } }2.2 消息处理核心逻辑实现消息流转的三大关键组件消息生产者接收业务系统请求将消息投递到RabbitMQService public class MessageProducer { Autowired private RabbitTemplate rabbitTemplate; public void pushToQueue(PushMessage message) { rabbitTemplate.convertAndSend( message.exchange, message.routingKey, message, m - { m.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); return m; }); } }消息消费者从队列获取消息并推送给在线用户RabbitListener(queues message.queue) public void handleMessage(PushMessage message) { String userId message.getTargetUserId(); if (onlineUsers.contains(userId)) { // 实时推送 webSocketHandler.pushMessage(userId, message); } else { // 存储离线消息 redisTemplate.opsForList().rightPush( offline: userId, JSON.toJSONString(message) ); } }连接管理器维护WebSocket连接状态public class WebSocketHandler extends TextWebSocketHandler { private static final ConcurrentMapString, WebSocketSession sessions new ConcurrentHashMap(); Override public void afterConnectionEstablished(WebSocketSession session) { String userId getUserIdFromSession(session); sessions.put(userId, session); pushOfflineMessages(userId); // 推送积压消息 } }3. 高并发优化策略3.1 RabbitMQ高级配置通过以下配置提升消息队列性能预取计数优化避免单个消费者过载Bean public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() { SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory(); factory.setPrefetchCount(100); // 每次获取100条消息 factory.setConcurrentConsumers(10); // 10个并发消费者 return factory; }队列镜像配置在RabbitMQ集群中启用镜像队列防止节点故障导致消息丢失spring.rabbitmq.addressesrabbit1:5672,rabbit2:5672,rabbit3:5672 spring.rabbitmq.publisher-confirmstrue spring.rabbitmq.publisher-returnstrue3.2 WebSocket集群方案单机WebSocket服务无法满足高并发需求需要实现会话共享通过Redis存储连接信息public class RedisSessionStore { public void storeSession(String userId, String serverId) { redisTemplate.opsForValue().set( ws: userId, serverId, 30, TimeUnit.MINUTES); } public String getSessionServer(String userId) { return redisTemplate.opsForValue().get(ws: userId); } }消息路由使用Nginx的sticky模块实现会话保持upstream websocket_cluster { sticky; server ws1:8080; server ws2:8080; server ws3:8080; } location /push { proxy_pass http://websocket_cluster; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_read_timeout 600s; }4. 生产环境注意事项4.1 连接保活机制实现完整的心跳检测流程客户端每25秒发送ping帧const heartbeatInterval 25000; let heartbeatTimer; function setupWebSocket() { const ws new WebSocket(wss://example.com/push); ws.onopen () { heartbeatTimer setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(__ping__); } }, heartbeatInterval); }; }服务端检测超时连接public class ConnectionMonitor { Scheduled(fixedRate 30000) public void checkAlive() { sessions.forEach((userId, session) - { if (System.currentTimeMillis() - session.getLastActiveTime() 40000) { session.close(); // 关闭超时连接 } }); } }4.2 异常处理与监控关键监控指标及处理方案指标阈值处理措施连接数80%最大负载自动扩容消息积压10万条增加消费者平均延迟500ms优化路由配置Prometheus监控示例- job_name: websocket metrics_path: /actuator/prometheus static_configs: - targets: [ws1:8080, ws2:8080]在电商大促期间这套系统成功支撑了峰值超过50万/秒的消息推送RabbitMQ队列积压控制在1万条以内WebSocket连接成功率保持在99.99%以上。实际部署时建议采用渐进式扩容策略先通过压力测试确定单节点容量再按需增加集群节点。