1. 分布式WebSocket的核心挑战与Redis解法当我们需要在分布式环境中实现WebSocket时最头疼的问题就是会话状态共享。想象一下这样的场景用户A通过服务器1建立了WebSocket连接但下次请求被负载均衡到了服务器2这时候服务器2根本不知道这个连接的存在。传统单机WebSocket的方案在这里完全失效。Redis作为内存数据库其pub/sub功能和数据结构特性恰好能解决这个痛点。我去年在电商秒杀系统的实时通知模块中就采用了这个方案成功支撑了5W的并发连接。具体来说Redis帮我们实现了连接路由表用Hash结构存储所有活跃连接的元信息用户ID→服务器节点消息中转站通过Pub/Sub通道广播消息到所有节点心跳监测用Sorted Set实现连接健康检查2. 架构设计与核心组件2.1 整体架构图[Client] ←WebSocket→ [Node1] ↔ Redis [Client] ←WebSocket→ [Node2] ↔ Redis [Client] ←WebSocket→ [Node3] ↔ Redis2.2 关键数据结构设计// 连接注册表 key: ws:connections:{userId} value: { node: node1_ip:port, createTime: 20230815T143022, lastActive: 20230815T150000 } // 全局消息通道 channel: ws:message:{groupId} // 心跳检测 key: ws:heartbeat value: ZSET(userId, timestamp)3. SpringBoot实现详解3.1 基础环境搭建!-- pom.xml关键依赖 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency3.2 WebSocket配置增强Configuration EnableWebSocket public class WebSocketConfig implements WebSocketConfigurer { Autowired private RedisTemplateString, String redisTemplate; Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(distributedHandler(), /ws) .setAllowedOrigins(*) .addInterceptors(new AuthInterceptor()); } Bean public WebSocketHandler distributedHandler() { return new DistributedWebSocketHandler(redisTemplate); } }3.3 核心处理器实现public class DistributedWebSocketHandler extends TextWebSocketHandler { private final RedisTemplateString, String redisTemplate; private static final String CONNECTION_PREFIX ws:connections:; Override public void afterConnectionEstablished(WebSocketSession session) { String userId getUserId(session); // 注册连接信息 redisTemplate.opsForHash().put( CONNECTION_PREFIX userId, node, getServerNodeId() ); // 加入心跳检测 redisTemplate.opsForZSet().add( ws:heartbeat, userId, System.currentTimeMillis() ); } Override protected void handleTextMessage(WebSocketSession session, TextMessage message) { // 消息格式{to:user123, content:...} MessageDTO msg parseMessage(message); if(msg.isBroadcast()) { redisTemplate.convertAndSend(ws:message:all, msg.getContent()); } else { redisTemplate.convertAndSend(ws:message:msg.getTo(), msg.getContent()); } } }4. Redis关键操作优化4.1 连接注册的原子性保证使用Redis的SETNX命令防止重复注册Boolean success redisTemplate.opsForValue().setIfAbsent( ws:lock: userId, 1, 1, TimeUnit.SECONDS ); if(success) { // 执行注册操作 }4.2 消息广播的性能优化采用pipeline批量发送redisTemplate.executePipelined((RedisCallbackObject) connection - { for(String userId : targetUsers) { connection.publish( (ws:message:userId).getBytes(), message.getBytes() ); } return null; });4.3 心跳检测的扫表策略Scheduled(fixedRate 30000) public void checkHeartbeat() { long cutoff System.currentTimeMillis() - 60000; SetString timeoutUsers redisTemplate.opsForZSet() .rangeByScore(ws:heartbeat, 0, cutoff); timeoutUsers.forEach(userId - { // 清理过期连接 redisTemplate.delete(CONNECTION_PREFIX userId); redisTemplate.opsForZSet().remove(ws:heartbeat, userId); }); }5. 生产环境踩坑实录5.1 连接闪断问题现象移动端网络切换时频繁断连 解决方案// 增加重连补偿机制 Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { if(status.getCode() 1006) { // 网络异常断开保留注册信息5分钟 redisTemplate.expire( CONNECTION_PREFIX getUserId(session), 5, TimeUnit.MINUTES ); } }5.2 Redis内存暴涨现象连接数超过3W时Redis内存占用超过10G 优化方案压缩存储的元信息字段对心跳ZSet采用分片存储设置合理的TTL策略5.3 消息堆积问题当消费者节点宕机时会出现消息堆积// 解决方案增加消费者状态监控 Scheduled(fixedRate 10000) public void monitorConsumer() { MapObject, Object nodes redisTemplate.opsForHash() .entries(ws:nodes:status); // 检测离线节点并触发消息重分配 }6. 性能压测数据使用JMeter对100台服务器集群测试并发连接数平均延迟吞吐量Redis CPU使用率1W28ms3.2k/s12%5W53ms8.7k/s39%10W112ms14k/s68%关键调优参数# Redis连接池配置 spring.redis.lettuce.pool.max-active200 spring.redis.lettuce.pool.max-wait500ms # WebSocket线程配置 server.tomcat.max-threads200 server.tomcat.accept-count507. 扩展优化方向7.1 区域化部署优化// 根据用户IP自动选择最近区域 public String selectServerRegion(String ip) { // 调用GeoIP服务 String region geoIpService.lookup(ip); return redisTemplate.opsForValue().get( ws:best_node: region ); }7.2 消息协议优化采用Protobuf替代JSON// 配置自定义消息转换器 Override public void configureMessageConverters( ListMessageConverter converters) { converters.add(new ProtobufMessageConverter()); }7.3 智能降级策略当Redis出现异常时自动降级CircuitBreaker(fallbackMethod localCacheFallback) public void sendMessage(MessageDTO msg) { // 正常Redis操作 } private void localCacheFallback(MessageDTO msg) { // 使用本地缓存暂存消息 localQueue.add(msg); }在最近的一次618大促中这套方案成功支撑了峰值23W的在线连接。有个特别实用的技巧在Redis的pub/sub通道名前加上环境前缀如dev_/prod_可以完美解决多环境共用一个Redis时的消息串扰问题。
