diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/.flattened-pom.xml b/yudao-framework/yudao-spring-boot-starter-websocket/.flattened-pom.xml new file mode 100644 index 00000000..bb525b8d --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/.flattened-pom.xml @@ -0,0 +1,92 @@ + + + 4.0.0 + cn.iocoder.boot + yudao-spring-boot-starter-websocket + 2025.08-SNAPSHOT + yudao-spring-boot-starter-websocket + WebSocket 框架，支持多节点的广播 + https://github.com/YunaiV/ruoyi-vue-pro + + + cn.iocoder.boot + yudao-common + 2025.08-SNAPSHOT + compile + + + cn.iocoder.boot + yudao-spring-boot-starter-security + 2025.08-SNAPSHOT + provided + + + org.springframework.boot + spring-boot-starter-websocket + 3.4.5 + compile + + + cn.iocoder.boot + yudao-spring-boot-starter-mq + 2025.08-SNAPSHOT + compile + + + org.springframework.kafka + spring-kafka + 3.3.5 + compile + true + + + org.springframework.amqp + spring-rabbit + 3.2.5 + compile + true + + + org.apache.rocketmq + rocketmq-spring-boot-starter + 2.3.2 + compile + true + + + cn.iocoder.boot + yudao-spring-boot-starter-biz-tenant + 2025.08-SNAPSHOT + provided + + + + + huaweicloud + huawei + https://mirrors.huaweicloud.com/repository/maven/ + + + aliyunmaven + aliyun + https://maven.aliyun.com/repository/public + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/milestone + + + + false + + spring-snapshots + Spring Snapshots + https://repo.spring.io/snapshot + + + diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/pom.xml b/yudao-framework/yudao-spring-boot-starter-websocket/pom.xml new file mode 100644 index 00000000..b534f10d --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/pom.xml @@ -0,0 +1,73 @@ + + + + cn.iocoder.boot + yudao-framework + ${revision} + + 4.0.0 + yudao-spring-boot-starter-websocket + jar + + ${project.artifactId} + WebSocket 框架，支持多节点的广播 + https://github.com/YunaiV/ruoyi-vue-pro + + + + + cn.iocoder.boot + yudao-common + + + + + + cn.iocoder.boot + yudao-spring-boot-starter-security + provided + + + + org.springframework.boot + spring-boot-starter-websocket + + + + + cn.iocoder.boot + yudao-spring-boot-starter-mq + + + org.springframework.kafka + spring-kafka + true + + + org.springframework.amqp + spring-rabbit + true + + + org.apache.rocketmq + rocketmq-spring-boot-starter + true + + + + + + cn.iocoder.boot + yudao-spring-boot-starter-biz-tenant + provided + + + + \ No newline at end of file diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/config/WebSocketProperties.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/config/WebSocketProperties.java new file mode 100644 index 00000000..f1d84f7a --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/config/WebSocketProperties.java @@ -0,0 +1,34 @@ +package cn.iocoder.yudao.framework.websocket.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + +import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.NotNull; + +/** + * WebSocket 配置项 + * + * @author xingyu4j + */ +@ConfigurationProperties("yudao.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"; + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/config/YudaoWebSocketAutoConfiguration.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/config/YudaoWebSocketAutoConfiguration.java new file mode 100644 index 00000000..3aded887 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/config/YudaoWebSocketAutoConfiguration.java @@ -0,0 +1,183 @@ +package cn.iocoder.yudao.framework.websocket.config; + +import cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration; +import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; +import cn.iocoder.yudao.framework.websocket.core.handler.JsonWebSocketMessageHandler; +import cn.iocoder.yudao.framework.websocket.core.listener.WebSocketMessageListener; +import cn.iocoder.yudao.framework.websocket.core.security.LoginUserHandshakeInterceptor; +import cn.iocoder.yudao.framework.websocket.core.security.WebSocketAuthorizeRequestsCustomizer; +import cn.iocoder.yudao.framework.websocket.core.sender.kafka.KafkaWebSocketMessageConsumer; +import cn.iocoder.yudao.framework.websocket.core.sender.kafka.KafkaWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.local.LocalWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.rabbitmq.RabbitMQWebSocketMessageConsumer; +import cn.iocoder.yudao.framework.websocket.core.sender.rabbitmq.RabbitMQWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.redis.RedisWebSocketMessageConsumer; +import cn.iocoder.yudao.framework.websocket.core.sender.redis.RedisWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.rocketmq.RocketMQWebSocketMessageConsumer; +import cn.iocoder.yudao.framework.websocket.core.sender.rocketmq.RocketMQWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionHandlerDecorator; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManager; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManagerImpl; +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 xingyu4j + */ +@AutoConfiguration(before = YudaoRedisMQConsumerAutoConfiguration.class) // before YudaoRedisMQConsumerAutoConfiguration 的原因是，需要保证 RedisWebSocketMessageConsumer 先创建，才能创建 RedisMessageListenerContainer +@EnableWebSocket // 开启 websocket +@ConditionalOnProperty(prefix = "yudao.websocket", value = "enable", matchIfMissing = true) // 允许使用 yudao.websocket.enable=false 禁用 websocket +@EnableConfigurationProperties(WebSocketProperties.class) +public class YudaoWebSocketAutoConfiguration { + + @Bean + public WebSocketConfigurer webSocketConfigurer(HandshakeInterceptor[] handshakeInterceptors, + WebSocketHandler webSocketHandler, + WebSocketProperties webSocketProperties) { + return registry -> registry + // 添加 WebSocketHandler + .addHandler(webSocketHandler, webSocketProperties.getPath()) + .addInterceptors(handshakeInterceptors) + // 允许跨域，否则前端连接会直接断开 + .setAllowedOriginPatterns("*"); + } + + @Bean + public HandshakeInterceptor handshakeInterceptor() { + return new LoginUserHandshakeInterceptor(); + } + + @Bean + public WebSocketHandler webSocketHandler(WebSocketSessionManager sessionManager, + List> messageListeners) { + // 1. 创建 JsonWebSocketMessageHandler 对象，处理消息 + JsonWebSocketMessageHandler messageHandler = new JsonWebSocketMessageHandler(messageListeners); + // 2. 创建 WebSocketSessionHandlerDecorator 对象，处理连接 + return new WebSocketSessionHandlerDecorator(messageHandler, sessionManager); + } + + @Bean + public WebSocketSessionManager webSocketSessionManager() { + return new WebSocketSessionManagerImpl(); + } + + @Bean + public WebSocketAuthorizeRequestsCustomizer webSocketAuthorizeRequestsCustomizer(WebSocketProperties webSocketProperties) { + return new WebSocketAuthorizeRequestsCustomizer(webSocketProperties); + } + + // ==================== Sender 相关 ==================== + + @Configuration + @ConditionalOnProperty(prefix = "yudao.websocket", name = "sender-type", havingValue = "local") + public class LocalWebSocketMessageSenderConfiguration { + + @Bean + public LocalWebSocketMessageSender localWebSocketMessageSender(WebSocketSessionManager sessionManager) { + return new LocalWebSocketMessageSender(sessionManager); + } + + } + + @Configuration + @ConditionalOnProperty(prefix = "yudao.websocket", name = "sender-type", havingValue = "redis") + public class RedisWebSocketMessageSenderConfiguration { + + @Bean + public RedisWebSocketMessageSender redisWebSocketMessageSender(WebSocketSessionManager sessionManager, + RedisMQTemplate redisMQTemplate) { + return new RedisWebSocketMessageSender(sessionManager, redisMQTemplate); + } + + @Bean + public RedisWebSocketMessageConsumer redisWebSocketMessageConsumer( + RedisWebSocketMessageSender redisWebSocketMessageSender) { + return new RedisWebSocketMessageConsumer(redisWebSocketMessageSender); + } + + } + + @Configuration + @ConditionalOnProperty(prefix = "yudao.websocket", name = "sender-type", havingValue = "rocketmq") + public class RocketMQWebSocketMessageSenderConfiguration { + + @Bean + public RocketMQWebSocketMessageSender rocketMQWebSocketMessageSender( + WebSocketSessionManager sessionManager, RocketMQTemplate rocketMQTemplate, + @Value("${yudao.websocket.sender-rocketmq.topic}") String topic) { + return new RocketMQWebSocketMessageSender(sessionManager, rocketMQTemplate, topic); + } + + @Bean + public RocketMQWebSocketMessageConsumer rocketMQWebSocketMessageConsumer( + RocketMQWebSocketMessageSender rocketMQWebSocketMessageSender) { + return new RocketMQWebSocketMessageConsumer(rocketMQWebSocketMessageSender); + } + + } + + @Configuration + @ConditionalOnProperty(prefix = "yudao.websocket", name = "sender-type", havingValue = "rabbitmq") + public class RabbitMQWebSocketMessageSenderConfiguration { + + @Bean + public RabbitMQWebSocketMessageSender rabbitMQWebSocketMessageSender( + WebSocketSessionManager sessionManager, RabbitTemplate rabbitTemplate, + TopicExchange websocketTopicExchange) { + return new RabbitMQWebSocketMessageSender(sessionManager, rabbitTemplate, websocketTopicExchange); + } + + @Bean + public RabbitMQWebSocketMessageConsumer rabbitMQWebSocketMessageConsumer( + RabbitMQWebSocketMessageSender rabbitMQWebSocketMessageSender) { + return new RabbitMQWebSocketMessageConsumer(rabbitMQWebSocketMessageSender); + } + + /** + * 创建 Topic Exchange + */ + @Bean + public TopicExchange websocketTopicExchange(@Value("${yudao.websocket.sender-rabbitmq.exchange}") String exchange) { + return new TopicExchange(exchange, + true, // durable: 是否持久化 + false); // exclusive: 是否排它 + } + + } + + @Configuration + @ConditionalOnProperty(prefix = "yudao.websocket", name = "sender-type", havingValue = "kafka") + public class KafkaWebSocketMessageSenderConfiguration { + + @Bean + public KafkaWebSocketMessageSender kafkaWebSocketMessageSender( + WebSocketSessionManager sessionManager, KafkaTemplate kafkaTemplate, + @Value("${yudao.websocket.sender-kafka.topic}") String topic) { + return new KafkaWebSocketMessageSender(sessionManager, kafkaTemplate, topic); + } + + @Bean + public KafkaWebSocketMessageConsumer kafkaWebSocketMessageConsumer( + KafkaWebSocketMessageSender kafkaWebSocketMessageSender) { + return new KafkaWebSocketMessageConsumer(kafkaWebSocketMessageSender); + } + + } + +} \ No newline at end of file diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/handler/JsonWebSocketMessageHandler.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/handler/JsonWebSocketMessageHandler.java new file mode 100644 index 00000000..120f529c --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/handler/JsonWebSocketMessageHandler.java @@ -0,0 +1,83 @@ +package cn.iocoder.yudao.framework.websocket.core.handler; + +import cn.hutool.core.util.StrUtil; +import cn.hutool.core.util.TypeUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils; +import cn.iocoder.yudao.framework.websocket.core.listener.WebSocketMessageListener; +import cn.iocoder.yudao.framework.websocket.core.message.JsonWebSocketMessage; +import cn.iocoder.yudao.framework.websocket.core.util.WebSocketFrameworkUtils; +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 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 芋道源码 + */ +@Slf4j +public class JsonWebSocketMessageHandler extends TextWebSocketHandler { + + /** + * type 与 WebSocketMessageListener 的映射 + */ + private final Map> listeners = new HashMap<>(); + + @SuppressWarnings({"rawtypes", "unchecked"}) + public JsonWebSocketMessageHandler(List listenersList) { + listenersList.forEach((Consumer) + listener -> listeners.put(listener.getType(), listener)); + } + + @Override + protected 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 获得对应的 WebSocketMessageListener + WebSocketMessageListener 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()); + } + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/listener/WebSocketMessageListener.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/listener/WebSocketMessageListener.java new file mode 100644 index 00000000..f3a62cc3 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/listener/WebSocketMessageListener.java @@ -0,0 +1,31 @@ +package cn.iocoder.yudao.framework.websocket.core.listener; + +import cn.iocoder.yudao.framework.websocket.core.message.JsonWebSocketMessage; +import org.springframework.web.socket.WebSocketSession; + +/** + * WebSocket 消息监听器接口 + * + * 目的：前端发送消息给后端后，处理对应 {@link #getType()} 类型的消息 + * + * @param 泛型，消息类型 + */ +public interface WebSocketMessageListener { + + /** + * 处理消息 + * + * @param session Session + * @param message 消息 + */ + void onMessage(WebSocketSession session, T message); + + /** + * 获得消息类型 + * + * @see JsonWebSocketMessage#getType() + * @return 消息类型 + */ + String getType(); + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/message/JsonWebSocketMessage.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/message/JsonWebSocketMessage.java new file mode 100644 index 00000000..0a55cd69 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/message/JsonWebSocketMessage.java @@ -0,0 +1,29 @@ +package cn.iocoder.yudao.framework.websocket.core.message; + +import cn.iocoder.yudao.framework.websocket.core.listener.WebSocketMessageListener; +import lombok.Data; + +import java.io.Serializable; + +/** + * JSON 格式的 WebSocket 消息帧 + * + * @author 芋道源码 + */ +@Data +public class JsonWebSocketMessage implements Serializable { + + /** + * 消息类型 + * + * 目的：用于分发到对应的 {@link WebSocketMessageListener} 实现类 + */ + private String type; + /** + * 消息内容 + * + * 要求 JSON 对象 + */ + private String content; + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/security/LoginUserHandshakeInterceptor.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/security/LoginUserHandshakeInterceptor.java new file mode 100644 index 00000000..3a31825f --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/security/LoginUserHandshakeInterceptor.java @@ -0,0 +1,42 @@ +package cn.iocoder.yudao.framework.websocket.core.security; + +import cn.iocoder.yudao.framework.security.core.LoginUser; +import cn.iocoder.yudao.framework.security.core.filter.TokenAuthenticationFilter; +import cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils; +import cn.iocoder.yudao.framework.websocket.core.util.WebSocketFrameworkUtils; +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 芋道源码 + */ +public class LoginUserHandshakeInterceptor implements HandshakeInterceptor { + + @Override + public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, + WebSocketHandler wsHandler, Map attributes) { + LoginUser loginUser = SecurityFrameworkUtils.getLoginUser(); + if (loginUser != null) { + WebSocketFrameworkUtils.setLoginUser(loginUser, attributes); + } + return true; + } + + @Override + public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, + WebSocketHandler wsHandler, Exception exception) { + // do nothing + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/security/WebSocketAuthorizeRequestsCustomizer.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/security/WebSocketAuthorizeRequestsCustomizer.java new file mode 100644 index 00000000..f260dfc2 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/security/WebSocketAuthorizeRequestsCustomizer.java @@ -0,0 +1,24 @@ +package cn.iocoder.yudao.framework.websocket.core.security; + +import cn.iocoder.yudao.framework.security.config.AuthorizeRequestsCustomizer; +import cn.iocoder.yudao.framework.websocket.config.WebSocketProperties; +import lombok.RequiredArgsConstructor; +import org.springframework.security.config.annotation.web.builders.HttpSecurity; +import org.springframework.security.config.annotation.web.configurers.AuthorizeHttpRequestsConfigurer; + +/** + * WebSocket 的权限自定义 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +public class WebSocketAuthorizeRequestsCustomizer extends AuthorizeRequestsCustomizer { + + private final WebSocketProperties webSocketProperties; + + @Override + public void customize(AuthorizeHttpRequestsConfigurer.AuthorizationManagerRequestMatcherRegistry registry) { + registry.requestMatchers(webSocketProperties.getPath()).permitAll(); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/AbstractWebSocketMessageSender.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/AbstractWebSocketMessageSender.java new file mode 100644 index 00000000..9309fd2f --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/AbstractWebSocketMessageSender.java @@ -0,0 +1,106 @@ +package cn.iocoder.yudao.framework.websocket.core.sender; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.StrUtil; +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; +import cn.iocoder.yudao.framework.websocket.core.message.JsonWebSocketMessage; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManager; +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 芋道源码 + */ +@Slf4j +@RequiredArgsConstructor +public abstract class AbstractWebSocketMessageSender implements WebSocketMessageSender { + + private final WebSocketSessionManager sessionManager; + + @Override + public void send(Integer userType, Long userId, String messageType, String messageContent) { + send(null, userType, userId, messageType, messageContent); + } + + @Override + public void send(Integer userType, String messageType, String messageContent) { + send(null, userType, null, messageType, messageContent); + } + + @Override + public 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 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) sessionManager.getSessionList(userType, userId); + } else if (userType != null) { + sessions = (List) 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 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); + } + }); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/WebSocketMessageSender.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/WebSocketMessageSender.java new file mode 100644 index 00000000..9f75ad52 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/WebSocketMessageSender.java @@ -0,0 +1,52 @@ +package cn.iocoder.yudao.framework.websocket.core.sender; + +import cn.iocoder.yudao.framework.common.util.json.JsonUtils; + +/** + * WebSocket 消息的发送器接口 + * + * @author 芋道源码 + */ +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)); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessage.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessage.java new file mode 100644 index 00000000..5a4cf531 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessage.java @@ -0,0 +1,35 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.kafka; + +import lombok.Data; + +/** + * Kafka 广播 WebSocket 的消息 + * + * @author 芋道源码 + */ +@Data +public class KafkaWebSocketMessage { + + /** + * Session 编号 + */ + private String sessionId; + /** + * 用户类型 + */ + private Integer userType; + /** + * 用户编号 + */ + private Long userId; + + /** + * 消息类型 + */ + private String messageType; + /** + * 消息内容 + */ + private String messageContent; + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessageConsumer.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessageConsumer.java new file mode 100644 index 00000000..4b255b93 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessageConsumer.java @@ -0,0 +1,28 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.kafka; + +import lombok.RequiredArgsConstructor; +import org.springframework.amqp.rabbit.annotation.RabbitHandler; +import org.springframework.kafka.annotation.KafkaListener; + +/** + * {@link KafkaWebSocketMessage} 广播消息的消费者，真正把消息发送出去 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +public class KafkaWebSocketMessageConsumer { + + private final KafkaWebSocketMessageSender kafkaWebSocketMessageSender; + + @RabbitHandler + @KafkaListener( + topics = "${yudao.websocket.sender-kafka.topic}", + // 在 Group 上，使用 UUID 生成其后缀。这样，启动的 Consumer 的 Group 不同，以达到广播消费的目的 + groupId = "${yudao.websocket.sender-kafka.consumer-group}" + "-" + "#{T(java.util.UUID).randomUUID()}") + public void onMessage(KafkaWebSocketMessage message) { + kafkaWebSocketMessageSender.send(message.getSessionId(), + message.getUserType(), message.getUserId(), + message.getMessageType(), message.getMessageContent()); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessageSender.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessageSender.java new file mode 100644 index 00000000..47bb598a --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/kafka/KafkaWebSocketMessageSender.java @@ -0,0 +1,67 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.kafka; + +import cn.iocoder.yudao.framework.websocket.core.sender.AbstractWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManager; +import lombok.extern.slf4j.Slf4j; +import org.springframework.kafka.core.KafkaTemplate; + +import java.util.concurrent.ExecutionException; + +/** + * 基于 Kafka 的 {@link WebSocketMessageSender} 实现类 + * + * @author 芋道源码 + */ +@Slf4j +public class KafkaWebSocketMessageSender extends AbstractWebSocketMessageSender { + + private final KafkaTemplate kafkaTemplate; + + private final String topic; + + public KafkaWebSocketMessageSender(WebSocketSessionManager sessionManager, + KafkaTemplate kafkaTemplate, + String topic) { + super(sessionManager); + this.kafkaTemplate = kafkaTemplate; + this.topic = topic; + } + + @Override + public void send(Integer userType, Long userId, String messageType, String messageContent) { + sendKafkaMessage(null, userId, userType, messageType, messageContent); + } + + @Override + public void send(Integer userType, String messageType, String messageContent) { + sendKafkaMessage(null, null, userType, messageType, messageContent); + } + + @Override + public void send(String sessionId, String messageType, String messageContent) { + sendKafkaMessage(sessionId, null, null, messageType, messageContent); + } + + /** + * 通过 Kafka 广播消息 + * + * @param sessionId Session 编号 + * @param userId 用户编号 + * @param userType 用户类型 + * @param messageType 消息类型 + * @param messageContent 消息内容 + */ + private void sendKafkaMessage(String sessionId, Long userId, Integer userType, + String messageType, String messageContent) { + KafkaWebSocketMessage mqMessage = new KafkaWebSocketMessage() + .setSessionId(sessionId).setUserId(userId).setUserType(userType) + .setMessageType(messageType).setMessageContent(messageContent); + try { + kafkaTemplate.send(topic, mqMessage).get(); + } catch (InterruptedException | ExecutionException e) { + log.error("[sendKafkaMessage][发送消息({}) 到 Kafka 失败]", mqMessage, e); + } + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/local/LocalWebSocketMessageSender.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/local/LocalWebSocketMessageSender.java new file mode 100644 index 00000000..66640ef3 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/local/LocalWebSocketMessageSender.java @@ -0,0 +1,20 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.local; + +import cn.iocoder.yudao.framework.websocket.core.sender.AbstractWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManager; + +/** + * 本地的 {@link WebSocketMessageSender} 实现类 + * + * 注意：仅仅适合单机场景！！！ + * + * @author 芋道源码 + */ +public class LocalWebSocketMessageSender extends AbstractWebSocketMessageSender { + + public LocalWebSocketMessageSender(WebSocketSessionManager sessionManager) { + super(sessionManager); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessage.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessage.java new file mode 100644 index 00000000..80a4bc17 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessage.java @@ -0,0 +1,37 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.rabbitmq; + +import lombok.Data; + +import java.io.Serializable; + +/** + * RabbitMQ 广播 WebSocket 的消息 + * + * @author 芋道源码 + */ +@Data +public class RabbitMQWebSocketMessage implements Serializable { + + /** + * Session 编号 + */ + private String sessionId; + /** + * 用户类型 + */ + private Integer userType; + /** + * 用户编号 + */ + private Long userId; + + /** + * 消息类型 + */ + private String messageType; + /** + * 消息内容 + */ + private String messageContent; + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessageConsumer.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessageConsumer.java new file mode 100644 index 00000000..59e38242 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessageConsumer.java @@ -0,0 +1,39 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.rabbitmq; + +import lombok.RequiredArgsConstructor; +import org.springframework.amqp.core.ExchangeTypes; +import org.springframework.amqp.rabbit.annotation.*; + +/** + * {@link RabbitMQWebSocketMessage} 广播消息的消费者，真正把消息发送出去 + * + * @author 芋道源码 + */ +@RabbitListener( + bindings = @QueueBinding( + value = @Queue( + // 在 Queue 的名字上，使用 UUID 生成其后缀。这样，启动的 Consumer 的 Queue 不同，以达到广播消费的目的 + name = "${yudao.websocket.sender-rabbitmq.queue}" + "-" + "#{T(java.util.UUID).randomUUID()}", + // Consumer 关闭时，该队列就可以被自动删除了 + autoDelete = "true" + ), + exchange = @Exchange( + name = "${yudao.websocket.sender-rabbitmq.exchange}", + type = ExchangeTypes.TOPIC, + declare = "false" + ) + ) +) +@RequiredArgsConstructor +public class RabbitMQWebSocketMessageConsumer { + + private final RabbitMQWebSocketMessageSender rabbitMQWebSocketMessageSender; + + @RabbitHandler + public void onMessage(RabbitMQWebSocketMessage message) { + rabbitMQWebSocketMessageSender.send(message.getSessionId(), + message.getUserType(), message.getUserId(), + message.getMessageType(), message.getMessageContent()); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessageSender.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessageSender.java new file mode 100644 index 00000000..065a5d6b --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rabbitmq/RabbitMQWebSocketMessageSender.java @@ -0,0 +1,62 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.rabbitmq; + +import cn.iocoder.yudao.framework.websocket.core.sender.AbstractWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManager; +import lombok.extern.slf4j.Slf4j; +import org.springframework.amqp.core.TopicExchange; +import org.springframework.amqp.rabbit.core.RabbitTemplate; + +/** + * 基于 RabbitMQ 的 {@link WebSocketMessageSender} 实现类 + * + * @author 芋道源码 + */ +@Slf4j +public class RabbitMQWebSocketMessageSender extends AbstractWebSocketMessageSender { + + private final RabbitTemplate rabbitTemplate; + + private final TopicExchange topicExchange; + + public RabbitMQWebSocketMessageSender(WebSocketSessionManager sessionManager, + RabbitTemplate rabbitTemplate, + TopicExchange topicExchange) { + super(sessionManager); + this.rabbitTemplate = rabbitTemplate; + this.topicExchange = topicExchange; + } + + @Override + public void send(Integer userType, Long userId, String messageType, String messageContent) { + sendRabbitMQMessage(null, userId, userType, messageType, messageContent); + } + + @Override + public void send(Integer userType, String messageType, String messageContent) { + sendRabbitMQMessage(null, null, userType, messageType, messageContent); + } + + @Override + public void send(String sessionId, String messageType, String messageContent) { + sendRabbitMQMessage(sessionId, null, null, messageType, messageContent); + } + + /** + * 通过 RabbitMQ 广播消息 + * + * @param sessionId Session 编号 + * @param userId 用户编号 + * @param userType 用户类型 + * @param messageType 消息类型 + * @param messageContent 消息内容 + */ + private void sendRabbitMQMessage(String sessionId, Long userId, Integer userType, + String messageType, String messageContent) { + RabbitMQWebSocketMessage mqMessage = new RabbitMQWebSocketMessage() + .setSessionId(sessionId).setUserId(userId).setUserType(userType) + .setMessageType(messageType).setMessageContent(messageContent); + rabbitTemplate.convertAndSend(topicExchange.getName(), null, mqMessage); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessage.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessage.java new file mode 100644 index 00000000..fb9ea0ca --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessage.java @@ -0,0 +1,34 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.redis; + +import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractRedisChannelMessage; +import lombok.Data; + +/** + * Redis 广播 WebSocket 的消息 + */ +@Data +public class RedisWebSocketMessage extends AbstractRedisChannelMessage { + + /** + * Session 编号 + */ + private String sessionId; + /** + * 用户类型 + */ + private Integer userType; + /** + * 用户编号 + */ + private Long userId; + + /** + * 消息类型 + */ + private String messageType; + /** + * 消息内容 + */ + private String messageContent; + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessageConsumer.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessageConsumer.java new file mode 100644 index 00000000..abce0069 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessageConsumer.java @@ -0,0 +1,23 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.redis; + +import cn.iocoder.yudao.framework.mq.redis.core.pubsub.AbstractRedisChannelMessageListener; +import lombok.RequiredArgsConstructor; + +/** + * {@link RedisWebSocketMessage} 广播消息的消费者，真正把消息发送出去 + * + * @author 芋道源码 + */ +@RequiredArgsConstructor +public class RedisWebSocketMessageConsumer extends AbstractRedisChannelMessageListener { + + private final RedisWebSocketMessageSender redisWebSocketMessageSender; + + @Override + public void onMessage(RedisWebSocketMessage message) { + redisWebSocketMessageSender.send(message.getSessionId(), + message.getUserType(), message.getUserId(), + message.getMessageType(), message.getMessageContent()); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessageSender.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessageSender.java new file mode 100644 index 00000000..d6004ac6 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/redis/RedisWebSocketMessageSender.java @@ -0,0 +1,57 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.redis; + +import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate; +import cn.iocoder.yudao.framework.websocket.core.sender.AbstractWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManager; +import lombok.extern.slf4j.Slf4j; + +/** + * 基于 Redis 的 {@link WebSocketMessageSender} 实现类 + * + * @author 芋道源码 + */ +@Slf4j +public class RedisWebSocketMessageSender extends AbstractWebSocketMessageSender { + + private final RedisMQTemplate redisMQTemplate; + + public RedisWebSocketMessageSender(WebSocketSessionManager sessionManager, + RedisMQTemplate redisMQTemplate) { + super(sessionManager); + this.redisMQTemplate = redisMQTemplate; + } + + @Override + public void send(Integer userType, Long userId, String messageType, String messageContent) { + sendRedisMessage(null, userId, userType, messageType, messageContent); + } + + @Override + public void send(Integer userType, String messageType, String messageContent) { + sendRedisMessage(null, null, userType, messageType, messageContent); + } + + @Override + public void send(String sessionId, String messageType, String messageContent) { + sendRedisMessage(sessionId, null, null, messageType, messageContent); + } + + /** + * 通过 Redis 广播消息 + * + * @param sessionId Session 编号 + * @param userId 用户编号 + * @param userType 用户类型 + * @param messageType 消息类型 + * @param messageContent 消息内容 + */ + private void sendRedisMessage(String sessionId, Long userId, Integer userType, + String messageType, String messageContent) { + RedisWebSocketMessage mqMessage = new RedisWebSocketMessage() + .setSessionId(sessionId).setUserId(userId).setUserType(userType) + .setMessageType(messageType).setMessageContent(messageContent); + redisMQTemplate.send(mqMessage); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessage.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessage.java new file mode 100644 index 00000000..91570e3e --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessage.java @@ -0,0 +1,35 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.rocketmq; + +import lombok.Data; + +/** + * RocketMQ 广播 WebSocket 的消息 + * + * @author 芋道源码 + */ +@Data +public class RocketMQWebSocketMessage { + + /** + * Session 编号 + */ + private String sessionId; + /** + * 用户类型 + */ + private Integer userType; + /** + * 用户编号 + */ + private Long userId; + + /** + * 消息类型 + */ + private String messageType; + /** + * 消息内容 + */ + private String messageContent; + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessageConsumer.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessageConsumer.java new file mode 100644 index 00000000..ab2e2c4d --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessageConsumer.java @@ -0,0 +1,30 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.rocketmq; + +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 芋道源码 + */ +@RocketMQMessageListener( // 重点：添加 @RocketMQMessageListener 注解，声明消费的 topic + topic = "${yudao.websocket.sender-rocketmq.topic}", + consumerGroup = "${yudao.websocket.sender-rocketmq.consumer-group}", + messageModel = MessageModel.BROADCASTING // 设置为广播模式，保证每个实例都能收到消息 +) +@RequiredArgsConstructor +public class RocketMQWebSocketMessageConsumer implements RocketMQListener { + + private final RocketMQWebSocketMessageSender rocketMQWebSocketMessageSender; + + @Override + public void onMessage(RocketMQWebSocketMessage message) { + rocketMQWebSocketMessageSender.send(message.getSessionId(), + message.getUserType(), message.getUserId(), + message.getMessageType(), message.getMessageContent()); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessageSender.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessageSender.java new file mode 100644 index 00000000..ed059bac --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/sender/rocketmq/RocketMQWebSocketMessageSender.java @@ -0,0 +1,61 @@ +package cn.iocoder.yudao.framework.websocket.core.sender.rocketmq; + +import cn.iocoder.yudao.framework.websocket.core.sender.AbstractWebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.sender.WebSocketMessageSender; +import cn.iocoder.yudao.framework.websocket.core.session.WebSocketSessionManager; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.spring.core.RocketMQTemplate; + +/** + * 基于 RocketMQ 的 {@link WebSocketMessageSender} 实现类 + * + * @author 芋道源码 + */ +@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; + } + + @Override + public void send(Integer userType, Long userId, String messageType, String messageContent) { + sendRocketMQMessage(null, userId, userType, messageType, messageContent); + } + + @Override + public void send(Integer userType, String messageType, String messageContent) { + sendRocketMQMessage(null, null, userType, messageType, messageContent); + } + + @Override + public 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); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionHandlerDecorator.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionHandlerDecorator.java new file mode 100644 index 00000000..600a4dd9 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionHandlerDecorator.java @@ -0,0 +1,49 @@ +package cn.iocoder.yudao.framework.websocket.core.session; + +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 芋道源码 + */ +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; + } + + @Override + public void afterConnectionEstablished(WebSocketSession session) { + // 实现 session 支持并发，可参考 https://blog.csdn.net/abu935009066/article/details/131218149 + session = new ConcurrentWebSocketSessionDecorator(session, SEND_TIME_LIMIT, BUFFER_SIZE_LIMIT); + // 添加到 WebSocketSessionManager 中 + sessionManager.addSession(session); + } + + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) { + sessionManager.removeSession(session); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionManager.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionManager.java new file mode 100644 index 00000000..ad1de23c --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionManager.java @@ -0,0 +1,53 @@ +package cn.iocoder.yudao.framework.websocket.core.session; + +import org.springframework.web.socket.WebSocketSession; + +import java.util.Collection; + +/** + * {@link WebSocketSession} 管理器的接口 + * + * @author 芋道源码 + */ +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 getSessionList(Integer userType); + + /** + * 获得指定用户编号的 Session 列表 + * + * @param userType 用户类型 + * @param userId 用户编号 + * @return Session 列表 + */ + Collection getSessionList(Integer userType, Long userId); + +} \ No newline at end of file diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionManagerImpl.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionManagerImpl.java new file mode 100644 index 00000000..6dba8985 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/session/WebSocketSessionManagerImpl.java @@ -0,0 +1,125 @@ +package cn.iocoder.yudao.framework.websocket.core.session; + +import cn.hutool.core.collection.CollUtil; +import cn.iocoder.yudao.framework.security.core.LoginUser; +import cn.iocoder.yudao.framework.tenant.core.context.TenantContextHolder; +import cn.iocoder.yudao.framework.websocket.core.util.WebSocketFrameworkUtils; +import org.springframework.web.socket.WebSocketSession; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * 默认的 {@link WebSocketSessionManager} 实现类 + * + * @author 芋道源码 + */ +public class WebSocketSessionManagerImpl implements WebSocketSessionManager { + + /** + * id 与 WebSocketSession 映射 + * + * key：Session 编号 + */ + private final ConcurrentMap idSessions = new ConcurrentHashMap<>(); + + /** + * user 与 WebSocketSession 映射 + * + * key1：用户类型 + * key2：用户编号 + */ + private final ConcurrentMap>> userSessions + = new ConcurrentHashMap<>(); + + @Override + public void addSession(WebSocketSession session) { + // 添加到 idSessions 中 + idSessions.put(session.getId(), session); + // 添加到 userSessions 中 + LoginUser user = WebSocketFrameworkUtils.getLoginUser(session); + if (user == null) { + return; + } + ConcurrentMap> userSessionsMap = userSessions.get(user.getUserType()); + if (userSessionsMap == null) { + userSessionsMap = new ConcurrentHashMap<>(); + if (userSessions.putIfAbsent(user.getUserType(), userSessionsMap) != null) { + userSessionsMap = userSessions.get(user.getUserType()); + } + } + CopyOnWriteArrayList sessions = userSessionsMap.get(user.getId()); + if (sessions == null) { + sessions = new CopyOnWriteArrayList<>(); + if (userSessionsMap.putIfAbsent(user.getId(), sessions) != null) { + sessions = userSessionsMap.get(user.getId()); + } + } + sessions.add(session); + } + + @Override + public void removeSession(WebSocketSession session) { + // 移除从 idSessions 中 + idSessions.remove(session.getId()); + // 移除从 idSessions 中 + LoginUser user = WebSocketFrameworkUtils.getLoginUser(session); + if (user == null) { + return; + } + ConcurrentMap> userSessionsMap = userSessions.get(user.getUserType()); + if (userSessionsMap == null) { + return; + } + CopyOnWriteArrayList sessions = userSessionsMap.get(user.getId()); + sessions.removeIf(session0 -> session0.getId().equals(session.getId())); + if (CollUtil.isEmpty(sessions)) { + userSessionsMap.remove(user.getId(), sessions); + } + } + + @Override + public WebSocketSession getSession(String id) { + return idSessions.get(id); + } + + @Override + public Collection getSessionList(Integer userType) { + ConcurrentMap> userSessionsMap = userSessions.get(userType); + if (CollUtil.isEmpty(userSessionsMap)) { + return new ArrayList<>(); + } + LinkedList result = new LinkedList<>(); // 避免扩容 + Long contextTenantId = TenantContextHolder.getTenantId(); + for (List sessions : userSessionsMap.values()) { + if (CollUtil.isEmpty(sessions)) { + continue; + } + // 特殊：如果租户不匹配，则直接排除 + if (contextTenantId != null) { + Long userTenantId = WebSocketFrameworkUtils.getTenantId(sessions.get(0)); + if (!contextTenantId.equals(userTenantId)) { + continue; + } + } + result.addAll(sessions); + } + return result; + } + + @Override + public Collection getSessionList(Integer userType, Long userId) { + ConcurrentMap> userSessionsMap = userSessions.get(userType); + if (CollUtil.isEmpty(userSessionsMap)) { + return new ArrayList<>(); + } + CopyOnWriteArrayList sessions = userSessionsMap.get(userId); + return CollUtil.isNotEmpty(sessions) ? new ArrayList<>(sessions) : new ArrayList<>(); + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/util/WebSocketFrameworkUtils.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/util/WebSocketFrameworkUtils.java new file mode 100644 index 00000000..58cdedc2 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/core/util/WebSocketFrameworkUtils.java @@ -0,0 +1,67 @@ +package cn.iocoder.yudao.framework.websocket.core.util; + +import cn.iocoder.yudao.framework.security.core.LoginUser; +import org.springframework.web.socket.WebSocketSession; + +import java.util.Map; + +/** + * 专属于 web 包的工具类 + * + * @author 芋道源码 + */ +public class WebSocketFrameworkUtils { + + public static final String ATTRIBUTE_LOGIN_USER = "LOGIN_USER"; + + /** + * 设置当前用户 + * + * @param loginUser 登录用户 + * @param attributes Session + */ + public static void setLoginUser(LoginUser loginUser, Map 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; + } + +} diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/package-info.java b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/package-info.java new file mode 100644 index 00000000..97bc5f95 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/java/cn/iocoder/yudao/framework/websocket/package-info.java @@ -0,0 +1,4 @@ +/** + * WebSocket 框架，支持多节点的广播 + */ +package cn.iocoder.yudao.framework.websocket; diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 00000000..6260e407 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1 @@ +cn.iocoder.yudao.framework.websocket.config.YudaoWebSocketAutoConfiguration \ No newline at end of file diff --git a/yudao-framework/yudao-spring-boot-starter-websocket/《芋道 Spring Boot WebSocket 入门》.md b/yudao-framework/yudao-spring-boot-starter-websocket/《芋道 Spring Boot WebSocket 入门》.md new file mode 100644 index 00000000..8df5a775 --- /dev/null +++ b/yudao-framework/yudao-spring-boot-starter-websocket/《芋道 Spring Boot WebSocket 入门》.md @@ -0,0 +1 @@ +