从零实现分布式WebSocket组件：设计模式深度实践指南

一、为什么需要WebSocket组件？

1. 实时通信需求
   • 传统HTTP轮询效率低，WebSocket提供全双工通信
   • 适用于即时聊天、实时数据监控、协同编辑等场景
2. 分布式系统挑战
   • 多节点部署时需解决会话同步问题
   • 跨节点消息广播需借助中间件（Redis/RocketMQ等）
3. 统一管理诉求
   • 会话管理、安全认证、消息路由等共性功能需抽象

二、组件核心功能全景

功能模块	实现要点
会话管理	按用户维度存储WebSocketSession，支持按ID/用户类型/用户ID查询
消息路由	根据消息类型分发到对应处理器（WebSocketMessageListener）
消息广播	支持本地/Redis/RocketMQ等多种发送策略，适配不同集群规模
安全认证	握手阶段Token验证，用户信息绑定到Session
并发控制	使用ConcurrentWebSocketSessionDecorator防止并发发送异常

三、组件工作全流程图解

1. 连接建立流程

2. 消息接收处理流程

3.消息发送流程（以RocketMQ为例）

四、四步构建WebSocket核心链路

步骤一：连接建立（认证与会话管理）

相关类

WebSocketFrameworkUtils

import org.springframework.web.socket.WebSocketSession;import java.util.Map;/*** 专属于 web 包的工具类** @author dyh*/
public class WebSocketFrameworkUtils {public static final String ATTRIBUTE_LOGIN_USER = "LOGIN_USER";/*** 设置当前用户** @param loginUser 登录用户* @param attributes Session*/public static void setLoginUser(LoginUser loginUser, Map<String, Object> attributes) {attributes.put(ATTRIBUTE_LOGIN_USER, loginUser);}/*** 获取当前用户** @return 当前用户*/public static LoginUser getLoginUser(WebSocketSession session) {return (LoginUser) session.getAttributes().get(ATTRIBUTE_LOGIN_USER);}/*** 获得当前用户的编号** @return 用户编号*/public static Long getLoginUserId(WebSocketSession session) {LoginUser loginUser = getLoginUser(session);return loginUser != null ? loginUser.getId() : null;}/*** 获得当前用户的类型** @return 用户编号*/public static Integer getLoginUserType(WebSocketSession session) {LoginUser loginUser = getLoginUser(session);return loginUser != null ? loginUser.getUserType() : null;}/*** 获得当前用户的租户编号** @param session Session* @return 租户编号*/public static Long getTenantId(WebSocketSession session) {LoginUser loginUser = getLoginUser(session);return loginUser != null ? loginUser.getTenantId() : null;}}

JsonWebSocketMessage

/*** JSON 格式的 WebSocket 消息帧** @author dyh*/
@Data
public class JsonWebSocketMessage implements Serializable {/*** 消息类型** 目的：用于分发到对应的 {@link WebSocketMessageListener} 实现类*/private String type;/*** 消息内容** 要求 JSON 对象*/private String content;}

涉及核心类

实现步骤

1. 握手拦截器（认证用户）

LoginUserHandshakeInterceptor

import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.server.HandshakeInterceptor;
import java.util.Map;/*** 登录用户的 {@link HandshakeInterceptor} 实现类** 流程如下：* 1. 前端连接 websocket 时，会通过拼接 ?token={token} 到 ws:// 连接后，这样它可以被 {@link TokenAuthenticationFilter} 所认证通过* 2. {@link LoginUserHandshakeInterceptor} 负责把 {@link LoginUser} 添加到 {@link WebSocketSession} 中** @author dyh*/
public class LoginUserHandshakeInterceptor implements HandshakeInterceptor {@Overridepublic boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,WebSocketHandler wsHandler, Map<String, Object> attributes) {LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();if (loginUser != null) {WebSocketFrameworkUtils.setLoginUser(loginUser, attributes);}return true;}@Overridepublic void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,WebSocketHandler wsHandler, Exception exception) {// do nothing}}

2. 会话装饰器（线程安全增强）

WebSocketSessionHandlerDecorator

import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
import org.springframework.web.socket.handler.WebSocketHandlerDecorator;/*** {@link WebSocketHandler} 的装饰类，实现了以下功能：** 1. {@link WebSocketSession} 连接或关闭时，使用 {@link #sessionManager} 进行管理* 2. 封装 {@link WebSocketSession} 支持并发操作** @author dyh*/
public class WebSocketSessionHandlerDecorator extends WebSocketHandlerDecorator {/*** 发送时间的限制，单位：毫秒*/private static final Integer SEND_TIME_LIMIT = 1000 * 5;/*** 发送消息缓冲上线，单位：bytes*/private static final Integer BUFFER_SIZE_LIMIT = 1024 * 100;private final WebSocketSessionManager sessionManager;public WebSocketSessionHandlerDecorator(WebSocketHandler delegate,WebSocketSessionManager sessionManager) {super(delegate);this.sessionManager = sessionManager;}@Overridepublic void afterConnectionEstablished(WebSocketSession session) {// 实现 session 支持并发session = new ConcurrentWebSocketSessionDecorator(session, SEND_TIME_LIMIT, BUFFER_SIZE_LIMIT);// 添加到 WebSocketSessionManager 中sessionManager.addSession(session);}@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) {sessionManager.removeSession(session);}}

3. 会话管理器（存储结构）

WebSocketSessionManager

import org.springframework.web.socket.WebSocketSession;import java.util.Collection;/*** {@link WebSocketSession} 管理器的接口** @author dyh*/
public interface WebSocketSessionManager {/*** 添加 Session** @param session Session*/void addSession(WebSocketSession session);/*** 移除 Session** @param session Session*/void removeSession(WebSocketSession session);/*** 获得指定编号的 Session** @param id Session 编号* @return Session*/WebSocketSession getSession(String id);/*** 获得指定用户类型的 Session 列表** @param userType 用户类型* @return Session 列表*/Collection<WebSocketSession> getSessionList(Integer userType);/*** 获得指定用户编号的 Session 列表** @param userType 用户类型* @param userId 用户编号* @return Session 列表*/Collection<WebSocketSession> getSessionList(Integer userType, Long userId);}

WebSocketSessionManagerImpl

import org.springframework.web.socket.WebSocketSession;
import cn.hutool.core.collection.CollUtil;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;/*** 默认的 {@link WebSocketSessionManager} 实现类* 负责管理WebSocket会话，包含双重存储结构：* 1. 基于会话ID的快速查找* 2. 基于用户类型+用户ID的分层存储* @author dyh*/
public class WebSocketSessionManagerImpl implements WebSocketSessionManager {/*** 会话ID与WebSocketSession的映射关系（线程安全）* Key格式：Session ID（WebSocket自动生成的唯一标识）* Value：对应的WebSocketSession对象*/private final ConcurrentMap<String, WebSocketSession> idSessions = new ConcurrentHashMap<>();/*** 用户维度存储WebSocketSession（三级嵌套线程安全结构）* Key1：用户类型（Integer，如1-管理员 2-普通用户）* Key2：用户ID（Long类型）* Value：该用户的所有WebSocket会话列表（线程安全的CopyOnWriteArrayList）*/private final ConcurrentMap<Integer, ConcurrentMap<Long, CopyOnWriteArrayList<WebSocketSession>>> userSessions = new ConcurrentHashMap<>();@Overridepublic void addSession(WebSocketSession session) {// 1. 加入ID索引idSessions.put(session.getId(), session);// 2. 加入用户维度索引LoginUser user = WebSocketFrameworkUtils.getLoginUser(session);if (user == null) return; // 未认证会话不记录用户维度// 2.1 获取或创建用户类型层（双重检查锁模式）userSessions.computeIfAbsent(user.getUserType(), k -> new ConcurrentHashMap<>())// 2.2 获取或创建用户ID层.computeIfAbsent(user.getId(), k -> new CopyOnWriteArrayList<>())// 2.3 添加会话到列表（线程安全操作）.add(session);}@Overridepublic void removeSession(WebSocketSession session) {// 1. 从ID索引移除idSessions.remove(session.getId());// 2. 从用户维度索引移除LoginUser user = WebSocketFrameworkUtils.getLoginUser(session);if (user == null) return;userSessions.computeIfPresent(user.getUserType(), (userType, userMap) -> {userMap.computeIfPresent(user.getId(), (userId, sessions) -> {// 2.1 移除指定会话（根据ID精准匹配）sessions.removeIf(s -> s.getId().equals(session.getId()));// 2.2 列表为空时自动清理用户ID层return sessions.isEmpty() ? null : sessions;});// 2.3 用户ID层空时自动清理用户类型层return userMap.isEmpty() ? null : userMap;});}@Overridepublic WebSocketSession getSession(String id) {return idSessions.get(id); // O(1)时间复杂度直接查找}@Overridepublic Collection<WebSocketSession> getSessionList(Integer userType) {// 1. 获取指定用户类型的所有会话映射ConcurrentMap<Long, CopyOnWriteArrayList<WebSocketSession>> userMap = userSessions.get(userType);if (CollUtil.isEmpty(userMap)) return Collections.emptyList();// 2. 多租户过滤处理Long currentTenantId = TenantContextHolder.getTenantId();LinkedList<WebSocketSession> result = new LinkedList<>(); // 避免ArrayList扩容开销userMap.values().forEach(sessions -> {if (CollUtil.isEmpty(sessions)) return;// 2.1 租户隔离检查（第一个会话代表用户所属租户）Long userTenantId = WebSocketFrameworkUtils.getTenantId(sessions.get(0));if (currentTenantId != null && !currentTenantId.equals(userTenantId)) return;result.addAll(sessions);});return result;}@Overridepublic Collection<WebSocketSession> getSessionList(Integer userType, Long userId) {// 直接获取指定用户的所有会话（带租户上下文过滤）return Optional.ofNullable(userSessions.get(userType)).map(userMap -> userMap.get(userId)).filter(CollUtil::isNotEmpty).map(ArrayList::new) // 返回拷贝避免直接操作内部集合.orElseGet(ArrayList::new);}
}

设计模式应用

• 装饰器模式：ConcurrentWebSocketSessionDecorator增强原生Session的并发能力
• 工厂方法模式：通过拦截器创建带有用户属性的Session对象

步骤二：接收消息（路由与处理）

涉及核心类

实现步骤

1. 消息处理器（路由分发）

JsonWebSocketMessageHandler

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import cn.hutool.core.util.StrUtil;
import cn.hutool.core.util.TypeUtil;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Consumer;/*** JSON 格式 {@link WebSocketHandler} 实现类** 基于 {@link JsonWebSocketMessage#getType()} 消息类型，调度到对应的 {@link WebSocketMessageListener} 监听器。** @author dyh*/
@Slf4j
public class JsonWebSocketMessageHandler extends TextWebSocketHandler {/*** type 与 WebSocketMessageListener 的映射*/private final Map<String, WebSocketMessageListener<Object>> listeners = new HashMap<>();@SuppressWarnings({"rawtypes", "unchecked"})public JsonWebSocketMessageHandler(List<? extends WebSocketMessageListener> listenersList) {listenersList.forEach((Consumer<WebSocketMessageListener>)listener -> listeners.put(listener.getType(), listener));}@Overrideprotected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {// 1.1 空消息，跳过if (message.getPayloadLength() == 0) {return;}// 1.2 ping 心跳消息，直接返回 pong 消息。if (message.getPayloadLength() == 4 && Objects.equals(message.getPayload(), "ping")) {session.sendMessage(new TextMessage("pong"));return;}// 2.1 解析消息try {JsonWebSocketMessage jsonMessage = JsonUtils.parseObject(message.getPayload(), JsonWebSocketMessage.class);if (jsonMessage == null) {log.error("[handleTextMessage][session({}) message({}) 解析为空]", session.getId(), message.getPayload());return;}if (StrUtil.isEmpty(jsonMessage.getType())) {log.error("[handleTextMessage][session({}) message({}) 类型为空]", session.getId(), message.getPayload());return;}// 2.2 获得对应的 WebSocketMessageListenerWebSocketMessageListener<Object> messageListener = listeners.get(jsonMessage.getType());if (messageListener == null) {log.error("[handleTextMessage][session({}) message({}) 监听器为空]", session.getId(), message.getPayload());return;}// 2.3 处理消息Type type = TypeUtil.getTypeArgument(messageListener.getClass(), 0);Object messageObj = JsonUtils.parseObject(jsonMessage.getContent(), type);Long tenantId = WebSocketFrameworkUtils.getTenantId(session);TenantUtils.execute(tenantId, () -> messageListener.onMessage(session, messageObj));} catch (Throwable ex) {log.error("[handleTextMessage][session({}) message({}) 处理异常]", session.getId(), message.getPayload());}}}

2. 定义监听器接口

WebSocketMessageListener

import org.springframework.web.socket.WebSocketSession;/*** WebSocket 消息监听器接口** 目的：前端发送消息给后端后，处理对应 {@link #getType()} 类型的消息** @param <T> 泛型，消息类型 * @author dyh          */
public interface WebSocketMessageListener<T> {/*** 处理消息** @param session Session* @param message 消息*/void onMessage(WebSocketSession session, T message);/*** 获得消息类型** @see JsonWebSocketMessage#getType()* @return 消息类型*/String getType();}

3. 业务监听器（使用示例）

注意：这里提供的是使用示例，具体接收到前端消息处理逻辑请根据业务实现
OrderMessageListener

   @Componentpublic class OrderMessageListener implements WebSocketMessageListener<OrderMessage> {@Overridepublic String getType() { return "order-msg"; }@Overridepublic void onMessage(WebSocketSession session, OrderMessage message) {orderService.process(message);}}

设计模式应用

• 策略模式：每个业务监听器对应一种消息处理策略
• 观察者模式：将消息类型与处理逻辑解耦，方便扩展新消息类型。

设计原则

• 依赖注入：Spring自动注入所有监听器实现
• 开闭原则：新增消息类型只需添加Listener实现类
• 单一职责：Handler只负责路由，Listener专注业务处理

步骤三：发送消息（策略化广播）

注意：此处只给了rocketmq消息策略，还可以自行实现redis，kafka，rabbitmq策略

相关类

RocketMQWebSocketMessage

import lombok.Data;/*** RocketMQ 广播 WebSocket 的消息** @author dyh*/
@Data
public class RocketMQWebSocketMessage {/*** Session 编号*/private String sessionId;/*** 用户类型*/private Integer userType;/*** 用户编号*/private Long userId;/*** 消息类型*/private String messageType;/*** 消息内容*/private String messageContent;}

涉及核心类

实现步骤

1. 定义发送器接口与抽象类

WebSocketMessageSender

/*** WebSocket 消息的发送器接口** @author dyh*/
public interface WebSocketMessageSender {/*** 发送消息给指定用户** @param userType 用户类型* @param userId 用户编号* @param messageType 消息类型* @param messageContent 消息内容，JSON 格式*/void send(Integer userType, Long userId, String messageType, String messageContent);/*** 发送消息给指定用户类型** @param userType 用户类型* @param messageType 消息类型* @param messageContent 消息内容，JSON 格式*/void send(Integer userType, String messageType, String messageContent);/*** 发送消息给指定 Session** @param sessionId Session 编号* @param messageType 消息类型* @param messageContent 消息内容，JSON 格式*/void send(String sessionId, String messageType, String messageContent);default void sendObject(Integer userType, Long userId, String messageType, Object messageContent) {send(userType, userId, messageType, JsonUtils.toJsonString(messageContent));}default void sendObject(Integer userType, String messageType, Object messageContent) {send(userType, messageType, JsonUtils.toJsonString(messageContent));}default void sendObject(String sessionId, String messageType, Object messageContent) {send(sessionId, messageType, JsonUtils.toJsonString(messageContent));}}

AbstractWebSocketMessageSender

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;/*** WebSocketMessageSender 实现类** @author dyh*/
@Slf4j
@RequiredArgsConstructor
public abstract class AbstractWebSocketMessageSender implements WebSocketMessageSender {private final WebSocketSessionManager sessionManager;@Overridepublic void send(Integer userType, Long userId, String messageType, String messageContent) {send(null, userType, userId, messageType, messageContent);}@Overridepublic void send(Integer userType, String messageType, String messageContent) {send(null, userType, null, messageType, messageContent);}@Overridepublic void send(String sessionId, String messageType, String messageContent) {send(sessionId, null, null, messageType, messageContent);}/*** 发送消息** @param sessionId Session 编号* @param userType 用户类型* @param userId 用户编号* @param messageType 消息类型* @param messageContent 消息内容*/public void send(String sessionId, Integer userType, Long userId, String messageType, String messageContent) {// 1. 获得 Session 列表List<WebSocketSession> sessions = Collections.emptyList();if (StrUtil.isNotEmpty(sessionId)) {WebSocketSession session = sessionManager.getSession(sessionId);if (session != null) {sessions = Collections.singletonList(session);}} else if (userType != null && userId != null) {sessions = (List<WebSocketSession>) sessionManager.getSessionList(userType, userId);} else if (userType != null) {sessions = (List<WebSocketSession>) sessionManager.getSessionList(userType);}if (CollUtil.isEmpty(sessions)) {if (log.isDebugEnabled()) {log.debug("[send][sessionId({}) userType({}) userId({}) messageType({}) messageContent({}) 未匹配到会话]",sessionId, userType, userId, messageType, messageContent);}}// 2. 执行发送doSend(sessions, messageType, messageContent);}/*** 发送消息的具体实现** @param sessions Session 列表* @param messageType 消息类型* @param messageContent 消息内容*/public void doSend(Collection<WebSocketSession> sessions, String messageType, String messageContent) {JsonWebSocketMessage message = new JsonWebSocketMessage().setType(messageType).setContent(messageContent);String payload = JsonUtils.toJsonString(message); // 关键，使用 JSON 序列化sessions.forEach(session -> {// 1. 各种校验，保证 Session 可以被发送if (session == null) {log.error("[doSend][session 为空, message({})]", message);return;}if (!session.isOpen()) {log.error("[doSend][session({}) 已关闭, message({})]", session.getId(), message);return;}// 2. 执行发送try {session.sendMessage(new TextMessage(payload));log.info("[doSend][session({}) 发送消息成功，message({})]", session.getId(), message);} catch (IOException ex) {log.error("[doSend][session({}) 发送消息失败，message({})]", session.getId(), message, ex);}});}}

2. 实现具体策略（以RocketMQ为例）

RocketMQWebSocketMessageSender

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;/*** 基于 RocketMQ 的 {@link WebSocketMessageSender} 实现类** @author dyh*/
@Slf4j
public class RocketMQWebSocketMessageSender extends AbstractWebSocketMessageSender {private final RocketMQTemplate rocketMQTemplate;private final String topic;public RocketMQWebSocketMessageSender(WebSocketSessionManager sessionManager,RocketMQTemplate rocketMQTemplate,String topic) {super(sessionManager);this.rocketMQTemplate = rocketMQTemplate;this.topic = topic;}@Overridepublic void send(Integer userType, Long userId, String messageType, String messageContent) {sendRocketMQMessage(null, userId, userType, messageType, messageContent);}@Overridepublic void send(Integer userType, String messageType, String messageContent) {sendRocketMQMessage(null, null, userType, messageType, messageContent);}@Overridepublic void send(String sessionId, String messageType, String messageContent) {sendRocketMQMessage(sessionId, null, null, messageType, messageContent);}/*** 通过 RocketMQ 广播消息** @param sessionId Session 编号* @param userId 用户编号* @param userType 用户类型* @param messageType 消息类型* @param messageContent 消息内容*/private void sendRocketMQMessage(String sessionId, Long userId, Integer userType,String messageType, String messageContent) {RocketMQWebSocketMessage mqMessage = new RocketMQWebSocketMessage().setSessionId(sessionId).setUserId(userId).setUserType(userType).setMessageType(messageType).setMessageContent(messageContent);rocketMQTemplate.syncSend(topic, mqMessage);}}

3. 消息消费者处理广播

RocketMQWebSocketMessageConsumer

import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;/*** {@link RocketMQWebSocketMessage} 广播消息的消费者，真正把消息发送出去** @author dyh*/
@RocketMQMessageListener( // 重点：添加 @RocketMQMessageListener 注解，声明消费的 topictopic = "${dyh.websocket.sender-rocketmq.topic}",consumerGroup = "${dyh.websocket.sender-rocketmq.consumer-group}",messageModel = MessageModel.BROADCASTING // 设置为广播模式，保证每个实例都能收到消息
)
@RequiredArgsConstructor
public class RocketMQWebSocketMessageConsumer implements RocketMQListener<RocketMQWebSocketMessage> {private final RocketMQWebSocketMessageSender rocketMQWebSocketMessageSender;@Overridepublic void onMessage(RocketMQWebSocketMessage message) {rocketMQWebSocketMessageSender.send(message.getSessionId(),message.getUserType(), message.getUserId(),message.getMessageType(), message.getMessageContent());}}

设计模式应用

• 策略模式：WebSocketMessageSender 接口统一发送策略，RedisWebSocketMessageSender 和 RocketMQWebSocketMessageSender 实现不同策略。
• 模板方法模式：AbstractWebSocketMessageSender

设计原则

• 开闭原则：新增发送策略只需继承AbstractWebSocketMessageSender抽象类，无需修改核心逻辑
• 依赖倒置：高层模块依赖WebSocketMessageSender抽象，不关心具体实现

步骤四：Spring 自动装配

相关类

WebSocketProperties

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;import javax.validation.constraints.NotEmpty;
import javax.validation.constraints.NotNull;/*** WebSocket 配置项** @author dyh*/
@ConfigurationProperties("dyh.websocket")
@Data
@Validated
public class WebSocketProperties {/*** WebSocket 的连接路径*/@NotEmpty(message = "WebSocket 的连接路径不能为空")private String path = "/ws";/*** 消息发送器的类型** 可选值：local、redis、rocketmq、kafka、rabbitmq*/@NotNull(message = "WebSocket 的消息发送者不能为空")private String senderType = "local";}

WebSocket 自动配置

DyhWebSocketAutoConfiguration

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.server.HandshakeInterceptor;import java.util.List;/*** WebSocket 自动配置** @author dyh*/
@AutoConfiguration(before = DyhRedisMQConsumerAutoConfiguration.class) // before DyhRedisMQConsumerAutoConfiguration 的原因是，需要保证 RedisWebSocketMessageConsumer 先创建，才能创建 RedisMessageListenerContainer
@EnableWebSocket // 开启 websocket
@ConditionalOnProperty(prefix = "dyh.websocket", value = "enable", matchIfMissing = true) // 允许使用 dyh.websocket.enable=false 禁用 websocket
@EnableConfigurationProperties(WebSocketProperties.class)
public class DyhWebSocketAutoConfiguration {@Beanpublic WebSocketConfigurer webSocketConfigurer(HandshakeInterceptor[] handshakeInterceptors,WebSocketHandler webSocketHandler,WebSocketProperties webSocketProperties) {return registry -> registry// 添加 WebSocketHandler.addHandler(webSocketHandler, webSocketProperties.getPath()).addInterceptors(handshakeInterceptors)// 允许跨域，否则前端连接会直接断开.setAllowedOriginPatterns("*");}@Beanpublic HandshakeInterceptor handshakeInterceptor() {return new LoginUserHandshakeInterceptor();}@Beanpublic WebSocketHandler webSocketHandler(WebSocketSessionManager sessionManager,List<? extends WebSocketMessageListener<?>> messageListeners) {// 1. 创建 JsonWebSocketMessageHandler 对象，处理消息JsonWebSocketMessageHandler messageHandler = new JsonWebSocketMessageHandler(messageListeners);// 2. 创建 WebSocketSessionHandlerDecorator 对象，处理连接return new WebSocketSessionHandlerDecorator(messageHandler, sessionManager);}@Beanpublic WebSocketSessionManager webSocketSessionManager() {return new WebSocketSessionManagerImpl();}@Beanpublic WebSocketAuthorizeRequestsCustomizer webSocketAuthorizeRequestsCustomizer(WebSocketProperties webSocketProperties) {return new WebSocketAuthorizeRequestsCustomizer(webSocketProperties);}// ==================== Sender 相关 ====================@Configuration@ConditionalOnProperty(prefix = "dyh.websocket", name = "sender-type", havingValue = "local")public class LocalWebSocketMessageSenderConfiguration {@Beanpublic LocalWebSocketMessageSender localWebSocketMessageSender(WebSocketSessionManager sessionManager) {return new LocalWebSocketMessageSender(sessionManager);}}@Configuration@ConditionalOnProperty(prefix = "dyh.websocket", name = "sender-type", havingValue = "redis")public class RedisWebSocketMessageSenderConfiguration {@Beanpublic RedisWebSocketMessageSender redisWebSocketMessageSender(WebSocketSessionManager sessionManager,RedisMQTemplate redisMQTemplate) {return new RedisWebSocketMessageSender(sessionManager, redisMQTemplate);}@Beanpublic RedisWebSocketMessageConsumer redisWebSocketMessageConsumer(RedisWebSocketMessageSender redisWebSocketMessageSender) {return new RedisWebSocketMessageConsumer(redisWebSocketMessageSender);}}@Configuration@ConditionalOnProperty(prefix = "dyh.websocket", name = "sender-type", havingValue = "rocketmq")public class RocketMQWebSocketMessageSenderConfiguration {@Beanpublic RocketMQWebSocketMessageSender rocketMQWebSocketMessageSender(WebSocketSessionManager sessionManager, RocketMQTemplate rocketMQTemplate,@Value("${dyh.websocket.sender-rocketmq.topic}") String topic) {return new RocketMQWebSocketMessageSender(sessionManager, rocketMQTemplate, topic);}@Beanpublic RocketMQWebSocketMessageConsumer rocketMQWebSocketMessageConsumer(RocketMQWebSocketMessageSender rocketMQWebSocketMessageSender) {return new RocketMQWebSocketMessageConsumer(rocketMQWebSocketMessageSender);}}@Configuration@ConditionalOnProperty(prefix = "dyh.websocket", name = "sender-type", havingValue = "rabbitmq")public class RabbitMQWebSocketMessageSenderConfiguration {@Beanpublic RabbitMQWebSocketMessageSender rabbitMQWebSocketMessageSender(WebSocketSessionManager sessionManager, RabbitTemplate rabbitTemplate,TopicExchange websocketTopicExchange) {return new RabbitMQWebSocketMessageSender(sessionManager, rabbitTemplate, websocketTopicExchange);}@Beanpublic RabbitMQWebSocketMessageConsumer rabbitMQWebSocketMessageConsumer(RabbitMQWebSocketMessageSender rabbitMQWebSocketMessageSender) {return new RabbitMQWebSocketMessageConsumer(rabbitMQWebSocketMessageSender);}/*** 创建 Topic Exchange*/@Beanpublic TopicExchange websocketTopicExchange(@Value("${dyh.websocket.sender-rabbitmq.exchange}") String exchange) {return new TopicExchange(exchange,true,  // durable: 是否持久化false);  // exclusive: 是否排它}}@Configuration@ConditionalOnProperty(prefix = "dyh.websocket", name = "sender-type", havingValue = "kafka")public class KafkaWebSocketMessageSenderConfiguration {@Beanpublic KafkaWebSocketMessageSender kafkaWebSocketMessageSender(WebSocketSessionManager sessionManager, KafkaTemplate<Object, Object> kafkaTemplate,@Value("${dyh.websocket.sender-kafka.topic}") String topic) {return new KafkaWebSocketMessageSender(sessionManager, kafkaTemplate, topic);}@Beanpublic KafkaWebSocketMessageConsumer kafkaWebSocketMessageConsumer(KafkaWebSocketMessageSender kafkaWebSocketMessageSender) {return new KafkaWebSocketMessageConsumer(kafkaWebSocketMessageSender);}}}

五、使用指南（含代码示例）

1. 发送消息示例

NotifyController

@RestController
public class NotifyController {@Autowiredprivate WebSocketMessageSender sender;// 发送给指定用户@PostMapping("/send")public void sendToUser(@RequestParam Long userId) {sender.sendObject(UserTypeEnum.ADMIN.getValue(), userId,"demo-message", new DemoMessage("Hello from server!"));}
}

2. 接收消息示例

DemoMessageListener

@Component
public class DemoMessageListener implements WebSocketMessageListener<DemoMessage> {@Overridepublic String getType() {return "demo-message"; // 声明处理的消息类型}@Overridepublic void onMessage(WebSocketSession session, DemoMessage message) {// 1. 获取当前用户Long userId = WebSocketFrameworkUtils.getLoginUserId(session);// 2. 处理业务逻辑System.out.println("收到来自用户：" + userId + " 的消息：" + message.getContent());}
}

六、设计模式与开发原则实践

1. 经典设计模式

模式名称	应用场景	代码示例
装饰器模式	增强WebSocketHandler的会话管理能力	WebSocketSessionHandlerDecorator
策略模式	可切换的消息发送策略（Local/Redis/RocketMQ）	WebSocketMessageSender多实现
观察者模式	消息类型与监听器的解耦	WebSocketMessageListener+消息路由
模板方法	定义消息发送流程骨架	AbstractWebSocketMessageSender#send()

2. 开发原则体现

• 单一职责原则：会话管理、消息路由、安全认证模块分离
• 开闭原则：新增消息类型只需实现新Listener，无需修改核心逻辑
• 依赖倒置：高层模块依赖WebSocketMessageSender抽象，不关心具体实现

七、从组件设计中学到的经验

1. 分层架构的价值
   • 将会话管理、消息路由、网络传输分层处理，提高可维护性
   • 通过抽象接口隔离变化点（如消息发送策略）
2. 分布式设计要点
   • 会话信息存储需考虑线程安全（ConcurrentHashMap）
   • 广播消息需借助中间件实现最终一致性
3. 扩展性设计技巧
   • 使用Spring条件装配实现策略动态切换
   • 监听器机制方便业务功能扩展
4. 生产环境考量
   • 心跳机制：内置Ping/Pong处理
   • 流量控制：ConcurrentWebSocketSessionDecorator限制缓冲区大小