diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketMessageSender.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/WebSocketMessageSender.java similarity index 56% rename from pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketMessageSender.java rename to pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/WebSocketMessageSender.java index 07ff60b97..001bfb6d9 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketMessageSender.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/WebSocketMessageSender.java @@ -1,38 +1,21 @@ -package com.pig4cloud.pigx.common.websocket.config; +package com.pig4cloud.pigx.common.websocket; import cn.hutool.json.JSONUtil; -import com.pig4cloud.pigx.common.websocket.holder.WebSocketSessionHolder; import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; 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; /** * @author Hccake 2021/1/4 * @version 1.0 */ @Slf4j -public class WebSocketMessageSender { +public final class WebSocketMessageSender { - public static void broadcast(String message) { - Collection sessions = WebSocketSessionHolder.getSessions(); - for (WebSocketSession session : sessions) { - send(session, message); - } - } - - public static boolean send(Object sessionKey, String message) { - WebSocketSession session = WebSocketSessionHolder.getSession(sessionKey); - if (session == null) { - log.info("[send] 当前 sessionKey：{} 对应 session 不在本服务中", sessionKey); - return false; - } - else { - return send(session, message); - } + private WebSocketMessageSender() { } public static void send(WebSocketSession session, JsonWebSocketMessage message) { diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfiguration.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfig.java similarity index 70% rename from pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfiguration.java rename to pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfig.java index baeedc499..b0ce1d614 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfiguration.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfig.java @@ -2,6 +2,8 @@ package com.pig4cloud.pigx.common.websocket.config; import com.pig4cloud.pigx.common.websocket.distribute.LocalMessageDistributor; import com.pig4cloud.pigx.common.websocket.distribute.MessageDistributor; +import com.pig4cloud.pigx.common.websocket.session.WebSocketSessionStore; +import lombok.RequiredArgsConstructor; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Bean; @@ -13,9 +15,12 @@ import org.springframework.context.annotation.Configuration; * @author hccake */ @ConditionalOnProperty(prefix = WebSocketProperties.PREFIX, name = "message-distributor", - havingValue = MessageDistributorTypeConstants.LOCAL) + havingValue = MessageDistributorTypeConstants.LOCAL, matchIfMissing = true) @Configuration(proxyBeanMethods = false) -public class LocalMessageDistributorConfiguration { +@RequiredArgsConstructor +public class LocalMessageDistributorConfig { + + private final WebSocketSessionStore webSocketSessionStore; /** * 本地基于内存的消息分发，不支持集群 @@ -24,7 +29,7 @@ public class LocalMessageDistributorConfiguration { @Bean @ConditionalOnMissingBean(MessageDistributor.class) public LocalMessageDistributor messageDistributor() { - return new LocalMessageDistributor(); + return new LocalMessageDistributor(webSocketSessionStore); } } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/MessageDistributorTypeConstants.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/MessageDistributorTypeConstants.java index 9e53ec911..dc397c3e4 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/MessageDistributorTypeConstants.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/MessageDistributorTypeConstants.java @@ -1,6 +1,8 @@ package com.pig4cloud.pigx.common.websocket.config; /** + * websocket 消息分发器类型常量类 + * * @author hccake */ public final class MessageDistributorTypeConstants { diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfiguration.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfig.java similarity index 59% rename from pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfiguration.java rename to pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfig.java index dc5f98efd..8cfca903d 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfiguration.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfig.java @@ -2,7 +2,8 @@ package com.pig4cloud.pigx.common.websocket.config; import com.pig4cloud.pigx.common.websocket.distribute.MessageDistributor; import com.pig4cloud.pigx.common.websocket.distribute.RedisMessageDistributor; -import com.pig4cloud.pigx.common.websocket.distribute.RedisWebsocketMessageListener; +import com.pig4cloud.pigx.common.websocket.distribute.RedisMessageListenerInitializer; +import com.pig4cloud.pigx.common.websocket.session.WebSocketSessionStore; import lombok.RequiredArgsConstructor; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; @@ -12,11 +13,8 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.StringRedisTemplate; -import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; -import javax.annotation.PostConstruct; - /** * 基于 Redis Pub/Sub 的消息分发器 * @@ -24,21 +22,17 @@ import javax.annotation.PostConstruct; */ @ConditionalOnClass(StringRedisTemplate.class) @ConditionalOnProperty(prefix = WebSocketProperties.PREFIX, name = "message-distributor", - havingValue = MessageDistributorTypeConstants.REDIS, matchIfMissing = true) + havingValue = MessageDistributorTypeConstants.REDIS) @Configuration(proxyBeanMethods = false) -public class RedisMessageDistributorConfiguration { +@RequiredArgsConstructor +public class RedisMessageDistributorConfig { + + private final WebSocketSessionStore webSocketSessionStore; @Bean @ConditionalOnMissingBean(MessageDistributor.class) public RedisMessageDistributor messageDistributor(StringRedisTemplate stringRedisTemplate) { - return new RedisMessageDistributor(stringRedisTemplate); - } - - @Bean - @ConditionalOnBean(RedisMessageDistributor.class) - @ConditionalOnMissingBean - public RedisWebsocketMessageListener redisWebsocketMessageListener(StringRedisTemplate stringRedisTemplate) { - return new RedisWebsocketMessageListener(stringRedisTemplate); + return new RedisMessageDistributor(webSocketSessionStore, stringRedisTemplate); } @Bean @@ -50,21 +44,12 @@ public class RedisMessageDistributorConfiguration { return container; } - @Configuration(proxyBeanMethods = false) - @ConditionalOnMissingBean(MessageDistributor.class) - @RequiredArgsConstructor - static class RedisMessageListenerRegisterConfiguration { - - private final RedisMessageListenerContainer redisMessageListenerContainer; - - private final RedisWebsocketMessageListener redisWebsocketMessageListener; - - @PostConstruct - public void addMessageListener() { - redisMessageListenerContainer.addMessageListener(redisWebsocketMessageListener, - new PatternTopic(RedisWebsocketMessageListener.CHANNEL)); - } - + @Bean + @ConditionalOnMissingBean + public RedisMessageListenerInitializer redisMessageListenerInitializer( + RedisMessageListenerContainer redisMessageListenerContainer, + RedisMessageDistributor redisWebsocketMessageListener) { + return new RedisMessageListenerInitializer(redisMessageListenerContainer, redisWebsocketMessageListener); } } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/SockJsServiceConfigurer.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/SockJsServiceConfigurer.java new file mode 100644 index 000000000..97a4bff8f --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/SockJsServiceConfigurer.java @@ -0,0 +1,18 @@ +package com.pig4cloud.pigx.common.websocket.config; + +import org.springframework.web.socket.config.annotation.SockJsServiceRegistration; + +/** + * SockJsService 配置类 + * + * @author hccake + */ +public interface SockJsServiceConfigurer { + + /** + * 配置 sockjs 相关 + * @param sockJsServiceRegistration sockJsService 注册类 + */ + void config(SockJsServiceRegistration sockJsServiceRegistration); + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketAutoConfiguration.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketAutoConfiguration.java index 7f819b32a..b917b2106 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketAutoConfiguration.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketAutoConfiguration.java @@ -1,50 +1,88 @@ package com.pig4cloud.pigx.common.websocket.config; import com.pig4cloud.pigx.common.websocket.handler.JsonMessageHandler; -import com.pig4cloud.pigx.common.websocket.holder.JsonMessageHandlerHolder; +import com.pig4cloud.pigx.common.websocket.handler.PingJsonMessageHandler; +import com.pig4cloud.pigx.common.websocket.holder.JsonMessageHandlerInitializer; +import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; import lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +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.Import; import org.springframework.web.socket.WebSocketHandler; import org.springframework.web.socket.config.annotation.EnableWebSocket; +import org.springframework.web.socket.config.annotation.SockJsServiceRegistration; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistration; import org.springframework.web.socket.server.HandshakeInterceptor; -import javax.annotation.PostConstruct; import java.util.List; /** * websocket自动配置 * - * @author Yakir + * @author Yakir Hccake */ -@Import(WebSocketHandlerConfig.class) +@Import({ WebSocketHandlerConfig.class, LocalMessageDistributorConfig.class, RedisMessageDistributorConfig.class }) @EnableWebSocket @RequiredArgsConstructor +@EnableConfigurationProperties(WebSocketProperties.class) public class WebSocketAutoConfiguration { private final WebSocketProperties webSocketProperties; - private final List jsonMessageHandlerList; - @Bean @ConditionalOnMissingBean public WebSocketConfigurer webSocketConfigurer(List handshakeInterceptor, - WebSocketHandler webSocketHandler) { - return registry -> registry.addHandler(webSocketHandler, webSocketProperties.getPath()) - .setAllowedOrigins(webSocketProperties.getAllowOrigins()) - .addInterceptors(handshakeInterceptor.toArray(new HandshakeInterceptor[0])); + WebSocketHandler webSocketHandler, + @Autowired(required = false) SockJsServiceConfigurer sockJsServiceConfigurer) { + return registry -> { + WebSocketHandlerRegistration registration = registry + .addHandler(webSocketHandler, webSocketProperties.getPath()) + .addInterceptors(handshakeInterceptor.toArray(new HandshakeInterceptor[0])); + + String[] allowedOrigins = webSocketProperties.getAllowedOrigins(); + if (allowedOrigins != null && allowedOrigins.length > 0) { + registration.setAllowedOrigins(allowedOrigins); + } + + String[] allowedOriginPatterns = webSocketProperties.getAllowedOriginPatterns(); + if (allowedOriginPatterns != null && allowedOriginPatterns.length > 0) { + registration.setAllowedOriginPatterns(allowedOriginPatterns); + } + + if (webSocketProperties.isWithSockjs()) { + SockJsServiceRegistration sockJsServiceRegistration = registration.withSockJS(); + if (sockJsServiceConfigurer != null) { + sockJsServiceConfigurer.config(sockJsServiceRegistration); + } + } + }; } /** - * 初始化时将所有的jsonMessageHandler注册到JsonMessageHandlerHolder中 + * 心跳处理器 + * @return PingJsonMessageHandler */ - @PostConstruct - public void initJsonMessageHandlerHolder() { - for (JsonMessageHandler jsonMessageHandler : jsonMessageHandlerList) { - JsonMessageHandlerHolder.addHandler(jsonMessageHandler); - } + @Bean + @ConditionalOnProperty(prefix = WebSocketProperties.PREFIX, name = "heartbeat", havingValue = "true", + matchIfMissing = true) + public PingJsonMessageHandler pingJsonMessageHandler() { + return new PingJsonMessageHandler(); + } + + /** + * 注册 JsonMessageHandlerInitializer 收集所有的 json 类型消息处理器 + * @param jsonMessageHandlerList json 类型消息处理器 + * @return JsonMessageHandlerInitializer + */ + @Bean + @ConditionalOnMissingBean + public JsonMessageHandlerInitializer jsonMessageHandlerInitializer( + List> jsonMessageHandlerList) { + return new JsonMessageHandlerInitializer(jsonMessageHandlerList); } } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java index 7215097d8..4fbe62e4c 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java @@ -1,78 +1,47 @@ package com.pig4cloud.pigx.common.websocket.config; -import com.pig4cloud.pigx.common.websocket.custom.PigxSessionKeyGenerator; -import com.pig4cloud.pigx.common.websocket.custom.UserAttributeHandshakeInterceptor; -import com.pig4cloud.pigx.common.websocket.handler.CustomPlanTextMessageHandler; import com.pig4cloud.pigx.common.websocket.handler.CustomWebSocketHandler; -import com.pig4cloud.pigx.common.websocket.handler.PingJsonMessageHandler; import com.pig4cloud.pigx.common.websocket.handler.PlanTextMessageHandler; -import com.pig4cloud.pigx.common.websocket.holder.MapSessionWebSocketHandlerDecorator; -import com.pig4cloud.pigx.common.websocket.holder.SessionKeyGenerator; +import com.pig4cloud.pigx.common.websocket.session.DefaultWebSocketSessionStore; +import com.pig4cloud.pigx.common.websocket.session.MapSessionWebSocketHandlerDecorator; +import com.pig4cloud.pigx.common.websocket.session.SessionKeyGenerator; +import com.pig4cloud.pigx.common.websocket.session.WebSocketSessionStore; import lombok.RequiredArgsConstructor; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.web.socket.WebSocketHandler; -import org.springframework.web.socket.handler.TextWebSocketHandler; -import org.springframework.web.socket.server.HandshakeInterceptor; /** * @author Hccake 2021/1/5 * @version 1.0 */ @RequiredArgsConstructor -@EnableConfigurationProperties(WebSocketProperties.class) public class WebSocketHandlerConfig { private final WebSocketProperties webSocketProperties; + /** + * WebSocket session 存储器 + * @return DefaultWebSocketSessionStore + */ @Bean - @ConditionalOnMissingBean(SessionKeyGenerator.class) - public SessionKeyGenerator sessionKeyGenerator() { - return new PigxSessionKeyGenerator(); + @ConditionalOnMissingBean + public WebSocketSessionStore webSocketSessionStore( + @Autowired(required = false) SessionKeyGenerator sessionKeyGenerator) { + return new DefaultWebSocketSessionStore(sessionKeyGenerator); } @Bean - public HandshakeInterceptor handshakeInterceptor() { - return new UserAttributeHandshakeInterceptor(); - } - - @Bean - @ConditionalOnMissingBean(PlanTextMessageHandler.class) - public PlanTextMessageHandler planTextMessageHandler() { - return new CustomPlanTextMessageHandler(); - } - - @Bean - @ConditionalOnMissingBean({ TextWebSocketHandler.class, PlanTextMessageHandler.class }) - public WebSocketHandler webSocketHandler1(@Autowired(required = false) SessionKeyGenerator sessionKeyGenerator) { - CustomWebSocketHandler customWebSocketHandler = new CustomWebSocketHandler(); - if (webSocketProperties.isMapSession()) { - return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator); - } - return customWebSocketHandler; - } - - @Bean - @ConditionalOnBean(PlanTextMessageHandler.class) - @ConditionalOnMissingBean(TextWebSocketHandler.class) - public WebSocketHandler webSocketHandler2(@Autowired(required = false) SessionKeyGenerator sessionKeyGenerator, - PlanTextMessageHandler planTextMessageHandler) { + @ConditionalOnMissingBean(WebSocketHandler.class) + public WebSocketHandler webSocketHandler(WebSocketSessionStore webSocketSessionStore, + @Autowired(required = false) PlanTextMessageHandler planTextMessageHandler) { CustomWebSocketHandler customWebSocketHandler = new CustomWebSocketHandler(planTextMessageHandler); if (webSocketProperties.isMapSession()) { - return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator); + return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, webSocketSessionStore, + webSocketProperties.getConcurrent()); } return customWebSocketHandler; } - @Bean - @ConditionalOnProperty(prefix = WebSocketProperties.PREFIX, name = "heartbeat", havingValue = "true", - matchIfMissing = true) - public PingJsonMessageHandler pingJsonMessageHandler() { - return new PingJsonMessageHandler(); - } - -} +} \ No newline at end of file diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java index 8bdc84cee..2b481daf4 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java @@ -1,7 +1,9 @@ package com.pig4cloud.pigx.common.websocket.config; +import com.pig4cloud.pigx.common.websocket.handler.ConcurrentWebSocketSessionOptions; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.NestedConfigurationProperty; /** * websocket props @@ -22,9 +24,14 @@ public class WebSocketProperties { private String path = "/ws/info"; /** - * 允许访问源 + * 允许跨域的源 */ - private String allowOrigins = "*"; + private String[] allowedOrigins; + + /** + * 允许跨域来源的匹配规则 + */ + private String[] allowedOriginPatterns = new String[] { "*" }; /** * 是否支持部分消息 @@ -41,9 +48,42 @@ public class WebSocketProperties { */ private boolean mapSession = true; + /** + * 是否开启 sockJs 支持 + */ + private boolean withSockjs = false; + + /** + * 多线程发送相关配置 + */ + @NestedConfigurationProperty + private ConcurrentWebSocketSessionOptions concurrent = new ConcurrentWebSocketSessionOptions(); + /** * 消息分发器：local | redis，默认 local, 如果自定义的话，可以配置为其他任意值 */ - private String messageDistributor = MessageDistributorTypeConstants.LOCAL; + private MessageDistributorTypeEnum messageDistributor = MessageDistributorTypeEnum.LOCAL; + + /** + * 消息分发器类型，用于解决集群场景下的消息跨节点推送问题 + */ + enum MessageDistributorTypeEnum { + + /** + * 本地缓存，不做跨节点分发 + */ + LOCAL, + + /** + * redis，利用 redis pub/sub 处理 + */ + REDIS, + + /** + * 自定义，用户自己实现一个 MessageDistributor + */ + CUSTOM; + + } } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java index a618b3ad6..b1b87af6a 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java @@ -3,6 +3,7 @@ package com.pig4cloud.pigx.common.websocket.custom; import com.pig4cloud.pigx.common.security.service.PigxUser; import com.pig4cloud.pigx.common.websocket.holder.SessionKeyGenerator; import lombok.RequiredArgsConstructor; +import org.springframework.cloud.client.ServiceInstance; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.WebSocketSession; diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/AbstractMessageDistributor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/AbstractMessageDistributor.java new file mode 100644 index 000000000..41b5d38d2 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/AbstractMessageDistributor.java @@ -0,0 +1,61 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +import cn.hutool.core.collection.CollectionUtil; +import com.pig4cloud.pigx.common.websocket.WebSocketMessageSender; +import com.pig4cloud.pigx.common.websocket.session.WebSocketSessionStore; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.socket.WebSocketSession; + +import java.util.Collection; + +/** + * @author hccake + */ +@Slf4j +public abstract class AbstractMessageDistributor implements MessageDistributor { + + private final WebSocketSessionStore webSocketSessionStore; + + protected AbstractMessageDistributor(WebSocketSessionStore webSocketSessionStore) { + this.webSocketSessionStore = webSocketSessionStore; + } + + @Override + public void doSend(MessageDO messageDO) { + + // 是否广播发送 + Boolean needBroadcast = messageDO.getNeedBroadcast(); + + // 获取待发送的 sessionKeys + Collection sessionKeys; + if (needBroadcast != null && needBroadcast) { + sessionKeys = webSocketSessionStore.getSessionKeys(); + } + else { + sessionKeys = messageDO.getSessionKeys(); + } + if (CollectionUtil.isEmpty(sessionKeys)) { + log.warn("发送 websocket 消息，确没有找到对应 sessionKeys, messageDo: {}", messageDO); + return; + } + + String messageText = messageDO.getMessageText(); + Boolean onlyOneClientInSameKey = messageDO.getOnlyOneClientInSameKey(); + + for (Object sessionKey : sessionKeys) { + Collection sessions = webSocketSessionStore.getSessions(sessionKey); + if (CollectionUtil.isNotEmpty(sessions)) { + // 相同 sessionKey 的客户端只推送一次操作 + if (onlyOneClientInSameKey != null && onlyOneClientInSameKey) { + WebSocketSession wsSession = CollectionUtil.get(sessions, 0); + WebSocketMessageSender.send(wsSession, messageText); + continue; + } + for (WebSocketSession wsSession : sessions) { + WebSocketMessageSender.send(wsSession, messageText); + } + } + } + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/LocalMessageDistributor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/LocalMessageDistributor.java index c5abdfd16..81c3aaadb 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/LocalMessageDistributor.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/LocalMessageDistributor.java @@ -1,12 +1,18 @@ package com.pig4cloud.pigx.common.websocket.distribute; +import com.pig4cloud.pigx.common.websocket.session.WebSocketSessionStore; + /** * 本地消息分发，直接进行发送 * * @author Hccake 2021/1/12 * @version 1.0 */ -public class LocalMessageDistributor implements MessageDistributor, MessageSender { +public class LocalMessageDistributor extends AbstractMessageDistributor { + + public LocalMessageDistributor(WebSocketSessionStore webSocketSessionStore) { + super(webSocketSessionStore); + } /** * 消息分发 diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDO.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDO.java index 5f9601bd9..98e86d1f8 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDO.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDO.java @@ -19,7 +19,12 @@ public class MessageDO { private Boolean needBroadcast; /** - * sessionKeys + * 对于拥有相同 sessionKey 的客户端，仅对其中的一个进行发送 + */ + private Boolean onlyOneClientInSameKey; + + /** + * 需要发送的 sessionKeys 集合，当广播时，不需要 */ private List sessionKeys; @@ -28,12 +33,4 @@ public class MessageDO { */ private String messageText; - /** - * 构建需要广播的message - * @author lingting 2021-03-25 17:28 - */ - public static MessageDO broadcastMessage(String text) { - return new MessageDO().setMessageText(text).setNeedBroadcast(true); - } - } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDistributor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDistributor.java index 0c7ccc693..865986211 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDistributor.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDistributor.java @@ -14,4 +14,10 @@ public interface MessageDistributor { */ void distribute(MessageDO messageDO); + /** + * 发送消息 + * @param messageDO 发送的消息 + */ + void doSend(MessageDO messageDO); + } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageSender.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageSender.java deleted file mode 100644 index f963f0352..000000000 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageSender.java +++ /dev/null @@ -1,34 +0,0 @@ -package com.pig4cloud.pigx.common.websocket.distribute; - -import cn.hutool.core.collection.CollectionUtil; -import com.pig4cloud.pigx.common.websocket.config.WebSocketMessageSender; - -import java.util.List; - -/** - * @author Hccake 2021/1/12 - * @version 1.0 - */ -public interface MessageSender { - - /** - * 发送消息 - * @param messageDO 发送的消息 - */ - default void doSend(MessageDO messageDO) { - Boolean needBroadcast = messageDO.getNeedBroadcast(); - String messageText = messageDO.getMessageText(); - List sessionKeys = messageDO.getSessionKeys(); - if (needBroadcast != null && needBroadcast) { - // 广播信息 - WebSocketMessageSender.broadcast(messageText); - } - else if (CollectionUtil.isNotEmpty(sessionKeys)) { - // 指定用户发送 - for (Object sessionKey : sessionKeys) { - WebSocketMessageSender.send(sessionKey, messageText); - } - } - } - -} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java index 380d9acab..55e28e436 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java @@ -1,20 +1,32 @@ package com.pig4cloud.pigx.common.websocket.distribute; import cn.hutool.json.JSONUtil; -import lombok.RequiredArgsConstructor; +import com.pig4cloud.pigx.common.websocket.session.WebSocketSessionStore; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializer; /** - * 消息分发器 + * 基于 redis PUB/SUB 的消息分发器, 订阅 websocket 发送消息，接收到消息时进行推送 * * @author Hccake 2021/1/12 * @version 1.0 */ -@RequiredArgsConstructor -public class RedisMessageDistributor implements MessageDistributor { +@Slf4j +public class RedisMessageDistributor extends AbstractMessageDistributor implements MessageListener { + + public static final String CHANNEL = "websocket-send"; private final StringRedisTemplate stringRedisTemplate; + public RedisMessageDistributor(WebSocketSessionStore webSocketSessionStore, + StringRedisTemplate stringRedisTemplate) { + super(webSocketSessionStore); + this.stringRedisTemplate = stringRedisTemplate; + } + /** * 消息分发 * @param messageDO 发送的消息 @@ -22,7 +34,23 @@ public class RedisMessageDistributor implements MessageDistributor { @Override public void distribute(MessageDO messageDO) { String str = JSONUtil.toJsonStr(messageDO); - stringRedisTemplate.convertAndSend(RedisWebsocketMessageListener.CHANNEL, str); + stringRedisTemplate.convertAndSend(CHANNEL, str); + } + + @Override + public void onMessage(Message message, byte[] bytes) { + log.info("redis channel Listener message send {}", message); + byte[] channelBytes = message.getChannel(); + RedisSerializer stringSerializer = stringRedisTemplate.getStringSerializer(); + String channel = stringSerializer.deserialize(channelBytes); + + // 这里没有使用通配符，所以一定是true + if (CHANNEL.equals(channel)) { + byte[] bodyBytes = message.getBody(); + String body = stringSerializer.deserialize(bodyBytes); + MessageDO messageDO = JSONUtil.toBean(body, MessageDO.class); + doSend(messageDO); + } } } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageListenerInitializer.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageListenerInitializer.java new file mode 100644 index 000000000..4892512f8 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageListenerInitializer.java @@ -0,0 +1,27 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +import lombok.RequiredArgsConstructor; +import org.springframework.data.redis.listener.PatternTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; + +import javax.annotation.PostConstruct; + +/** + * 初始化 redis 消息的监听器 + * + * @author Hccake + */ +@RequiredArgsConstructor +public class RedisMessageListenerInitializer { + + private final RedisMessageListenerContainer redisMessageListenerContainer; + + private final RedisMessageDistributor redisWebsocketMessageListener; + + @PostConstruct + public void addMessageListener() { + redisMessageListenerContainer.addMessageListener(redisWebsocketMessageListener, + new PatternTopic(RedisMessageDistributor.CHANNEL)); + } + +} \ No newline at end of file diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisWebsocketMessageListener.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisWebsocketMessageListener.java deleted file mode 100644 index 5b5be796a..000000000 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisWebsocketMessageListener.java +++ /dev/null @@ -1,41 +0,0 @@ -package com.pig4cloud.pigx.common.websocket.distribute; - -import cn.hutool.json.JSONUtil; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.data.redis.connection.Message; -import org.springframework.data.redis.connection.MessageListener; -import org.springframework.data.redis.core.StringRedisTemplate; -import org.springframework.data.redis.serializer.RedisSerializer; - -/** - * redis订阅 websocket 发送消息，接收到消息时进行推送 - * - * @author Hccake 2021/1/12 - * @version 1.0 - */ -@Slf4j -@RequiredArgsConstructor -public class RedisWebsocketMessageListener implements MessageListener, MessageSender { - - public static final String CHANNEL = "websocket-send"; - - private final StringRedisTemplate stringRedisTemplate; - - @Override - public void onMessage(Message message, byte[] bytes) { - log.info("redis channel Listener message send {}", message); - byte[] channelBytes = message.getChannel(); - RedisSerializer stringSerializer = stringRedisTemplate.getStringSerializer(); - String channel = stringSerializer.deserialize(channelBytes); - - // 这里没有使用通配符，所以一定是true - if (CHANNEL.equals(channel)) { - byte[] bodyBytes = message.getBody(); - String body = stringSerializer.deserialize(bodyBytes); - MessageDO messageDO = JSONUtil.toBean(body, MessageDO.class); - doSend(messageDO); - } - } - -} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/exception/ErrorJsonMessageException.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/exception/ErrorJsonMessageException.java new file mode 100644 index 000000000..f690a7cd0 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/exception/ErrorJsonMessageException.java @@ -0,0 +1,14 @@ +package com.pig4cloud.pigx.common.websocket.exception; + +/** + * 错误的 json 消息 + * + * @author hccake + */ +public class ErrorJsonMessageException extends RuntimeException { + + public ErrorJsonMessageException(String message) { + super(message); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/ConcurrentWebSocketSessionOptions.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/ConcurrentWebSocketSessionOptions.java new file mode 100644 index 000000000..486d60a11 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/ConcurrentWebSocketSessionOptions.java @@ -0,0 +1,34 @@ +package com.pig4cloud.pigx.common.websocket.handler; + +import lombok.Data; +import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator; + +/** + * 并发使用 WebSocketSession 的相关配置 + * + * @author hccake + */ +@Data +public class ConcurrentWebSocketSessionOptions { + + /** + * 是否在多线程环境下进行发送 + */ + private boolean enable = false; + + /** + * 发送时间的限制（ms） + */ + private int sendTimeLimit = 1000 * 5; + + /** + * 发送消息缓冲上限 (byte) + */ + private int bufferSizeLimit = 1024 * 100; + + /** + * 溢出时的执行策略 + */ + ConcurrentWebSocketSessionDecorator.OverflowStrategy overflowStrategy = ConcurrentWebSocketSessionDecorator.OverflowStrategy.TERMINATE; + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/CustomWebSocketHandler.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/CustomWebSocketHandler.java index 6748a22b9..af6936313 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/CustomWebSocketHandler.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/CustomWebSocketHandler.java @@ -4,6 +4,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.json.JsonReadFeature; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.pig4cloud.pigx.common.websocket.exception.ErrorJsonMessageException; import com.pig4cloud.pigx.common.websocket.holder.JsonMessageHandlerHolder; import com.pig4cloud.pigx.common.websocket.message.AbstractJsonWebSocketMessage; import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; @@ -36,18 +37,21 @@ public class CustomWebSocketHandler extends TextWebSocketHandler { } @Override - public void handleTextMessage(WebSocketSession session, TextMessage message) throws JsonProcessingException { + public void handleTextMessage(WebSocketSession session, TextMessage message) { // 空消息不处理 if (message.getPayloadLength() == 0) { return; } - // 消息类型必有一属性type，先解析，获取该属性 + // 获取消息载荷 String payload = message.getPayload(); - JsonNode jsonNode = MAPPER.readTree(payload); - JsonNode typeNode = jsonNode.get(AbstractJsonWebSocketMessage.TYPE_FIELD); - - if (typeNode == null) { + try { + // 尝试使用 json 处理 + handleWithJson(session, payload); + } + catch (ErrorJsonMessageException ex) { + log.debug("消息载荷 [{}] 回退使用 PlanTextMessageHandler，原因：{}", payload, ex.getMessage()); + // fallback 使用普通文本处理 if (planTextMessageHandler != null) { planTextMessageHandler.handle(session, payload); } @@ -55,19 +59,46 @@ public class CustomWebSocketHandler extends TextWebSocketHandler { log.error("[handleTextMessage] 普通文本消息（{}）没有对应的消息处理器", payload); } } - else { - String messageType = typeNode.asText(); - // 获得对应的消息处理器 - JsonMessageHandler jsonMessageHandler = JsonMessageHandlerHolder.getHandler(messageType); - if (jsonMessageHandler == null) { - log.error("[handleTextMessage] 消息类型（{}）不存在对应的消息处理器", messageType); - return; - } - // 消息处理 - Class messageClass = jsonMessageHandler.getMessageClass(); - JsonWebSocketMessage websocketMessageJson = MAPPER.treeToValue(jsonNode, messageClass); - jsonMessageHandler.handle(session, websocketMessageJson); + + } + + private void handleWithJson(WebSocketSession session, String payload) { + // 消息类型必有一属性type，先解析，获取该属性 + JsonNode jsonNode = null; + try { + jsonNode = MAPPER.readTree(payload); } + catch (JsonProcessingException e) { + throw new ErrorJsonMessageException("json 解析异常"); + } + + // 必须是 object 类型 + if (!jsonNode.isObject()) { + throw new ErrorJsonMessageException("json 格式异常！非 object 类型！"); + } + + JsonNode typeNode = jsonNode.get(AbstractJsonWebSocketMessage.TYPE_FIELD); + String messageType = typeNode.asText(); + if (messageType == null) { + throw new ErrorJsonMessageException("json 无 type 属性"); + } + + // 获得对应的消息处理器 + JsonMessageHandler jsonMessageHandler = JsonMessageHandlerHolder.getHandler(messageType); + if (jsonMessageHandler == null) { + log.error("[handleTextMessage] 消息类型（{}）不存在对应的消息处理器", messageType); + return; + } + // 消息处理 + Class messageClass = jsonMessageHandler.getMessageClass(); + JsonWebSocketMessage websocketMessageJson; + try { + websocketMessageJson = MAPPER.treeToValue(jsonNode, messageClass); + } + catch (JsonProcessingException e) { + throw new ErrorJsonMessageException("消息序列化异常，class " + messageClass); + } + jsonMessageHandler.handle(session, websocketMessageJson); } } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PingJsonMessageHandler.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PingJsonMessageHandler.java index aea06792b..550871b28 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PingJsonMessageHandler.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PingJsonMessageHandler.java @@ -1,6 +1,6 @@ package com.pig4cloud.pigx.common.websocket.handler; -import com.pig4cloud.pigx.common.websocket.config.WebSocketMessageSender; +import com.pig4cloud.pigx.common.websocket.WebSocketMessageSender; import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; import com.pig4cloud.pigx.common.websocket.message.PingJsonWebSocketMessage; import com.pig4cloud.pigx.common.websocket.message.PongJsonWebSocketMessage; diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerHolder.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerHolder.java index d19ea85bf..8fc8f845c 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerHolder.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerHolder.java @@ -1,6 +1,7 @@ package com.pig4cloud.pigx.common.websocket.holder; import com.pig4cloud.pigx.common.websocket.handler.JsonMessageHandler; +import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -14,13 +15,13 @@ public final class JsonMessageHandlerHolder { private JsonMessageHandlerHolder() { } - private static final Map MESSAGE_HANDLER_MAP = new ConcurrentHashMap<>(); + private static final Map> MESSAGE_HANDLER_MAP = new ConcurrentHashMap<>(); - public static JsonMessageHandler getHandler(String type) { + public static JsonMessageHandler getHandler(String type) { return MESSAGE_HANDLER_MAP.get(type); } - public static void addHandler(JsonMessageHandler jsonMessageHandler) { + public static void addHandler(JsonMessageHandler jsonMessageHandler) { MESSAGE_HANDLER_MAP.put(jsonMessageHandler.type(), jsonMessageHandler); } diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerInitializer.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerInitializer.java new file mode 100644 index 000000000..162cac0c3 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerInitializer.java @@ -0,0 +1,31 @@ +package com.pig4cloud.pigx.common.websocket.holder; + +import com.pig4cloud.pigx.common.websocket.handler.JsonMessageHandler; +import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; +import lombok.RequiredArgsConstructor; + +import javax.annotation.PostConstruct; +import java.util.List; + +/** + *

+ * JsonMessageHandler 初始化器 + *

+ * 将所有的 jsonMessageHandler 收集到 JsonMessageHandlerHolder 中 + * + * @author Hccake + */ +@RequiredArgsConstructor +public class JsonMessageHandlerInitializer { + + private final List> jsonMessageHandlerList; + + @SuppressWarnings("unchecked") + @PostConstruct + public void initJsonMessageHandlerHolder() { + for (JsonMessageHandler jsonMessageHandler : jsonMessageHandlerList) { + JsonMessageHandlerHolder.addHandler((JsonMessageHandler) jsonMessageHandler); + } + } + +} \ No newline at end of file diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java deleted file mode 100644 index 74c3ac43d..000000000 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java +++ /dev/null @@ -1,46 +0,0 @@ -package com.pig4cloud.pigx.common.websocket.holder; - -import org.springframework.web.socket.CloseStatus; -import org.springframework.web.socket.WebSocketHandler; -import org.springframework.web.socket.WebSocketSession; -import org.springframework.web.socket.handler.WebSocketHandlerDecorator; - -/** - * WebSocketHandler 装饰器，该装饰器主要用于在开启和关闭连接时，进行session的映射存储与释放 - * - * @author Hccake 2020/12/31 - * @version 1.0 - */ -public class MapSessionWebSocketHandlerDecorator extends WebSocketHandlerDecorator { - - private final SessionKeyGenerator sessionKeyGenerator; - - public MapSessionWebSocketHandlerDecorator(WebSocketHandler delegate, SessionKeyGenerator sessionKeyGenerator) { - super(delegate); - this.sessionKeyGenerator = sessionKeyGenerator; - } - - /** - * websocket 连接时执行的动作 - * @param session websocket session 对象 - * @throws Exception 异常对象 - */ - @Override - public void afterConnectionEstablished(final WebSocketSession session) throws Exception { - Object sessionKey = sessionKeyGenerator.sessionKey(session); - WebSocketSessionHolder.addSession(sessionKey, session); - } - - /** - * websocket 关闭连接时执行的动作 - * @param session websocket session 对象 - * @param closeStatus 关闭状态对象 - * @throws Exception 异常对象 - */ - @Override - public void afterConnectionClosed(final WebSocketSession session, CloseStatus closeStatus) throws Exception { - Object sessionKey = sessionKeyGenerator.sessionKey(session); - WebSocketSessionHolder.removeSession(sessionKey); - } - -} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/WebSocketSessionHolder.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/WebSocketSessionHolder.java deleted file mode 100644 index e06cc3c17..000000000 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/WebSocketSessionHolder.java +++ /dev/null @@ -1,65 +0,0 @@ -package com.pig4cloud.pigx.common.websocket.holder; - -import org.springframework.web.socket.WebSocketSession; - -import java.util.Collection; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; - -/** - * WebSocketSession 持有者 主要用于保存当前所有在线的会话信息 - * - * @author Hccake 2021/1/4 - * @version 1.0 - */ -public final class WebSocketSessionHolder { - - private WebSocketSessionHolder() { - } - - private static final Map USER_SESSION_MAP = new ConcurrentHashMap<>(); - - /** - * 添加一个 session - * @param sessionKey session 唯一标识 - * @param session 待添加的 WebSocketSession - */ - public static void addSession(Object sessionKey, WebSocketSession session) { - USER_SESSION_MAP.put(sessionKey, session); - } - - /** - * 删除一个 session - * @param sessionKey session唯一标识 - */ - public static void removeSession(Object sessionKey) { - USER_SESSION_MAP.remove(sessionKey); - } - - /** - * 获取指定标识的 session - * @param sessionKey session唯一标识 - * @return WebSocketSession 该标识对应的 session - */ - public static WebSocketSession getSession(Object sessionKey) { - return USER_SESSION_MAP.get(sessionKey); - } - - /** - * 获取当前所有在线的 session - * @return Collection session集合 - */ - public static Collection getSessions() { - return USER_SESSION_MAP.values(); - } - - /** - * 获取所有在线的用户标识 - * @return Set session唯一标识集合 - */ - public static Set getSessionKeys() { - return USER_SESSION_MAP.keySet(); - } - -} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/DefaultWebSocketSessionStore.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/DefaultWebSocketSessionStore.java new file mode 100644 index 000000000..2c32282a6 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/DefaultWebSocketSessionStore.java @@ -0,0 +1,96 @@ +package com.pig4cloud.pigx.common.websocket.session; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.socket.WebSocketSession; + +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +/** + * 默认的 WebSocketSession 存储器 + * + * @author hccake + */ +@Slf4j +public class DefaultWebSocketSessionStore implements WebSocketSessionStore { + + private final SessionKeyGenerator sessionKeyGenerator; + + private final ConcurrentHashMap> sessionKeyToWsSessions = new ConcurrentHashMap<>(); + + public DefaultWebSocketSessionStore(SessionKeyGenerator sessionKeyGenerator) { + this.sessionKeyGenerator = sessionKeyGenerator; + } + + /** + * 添加一个 wsSession + * @param wsSession 待添加的 WebSocketSession + */ + @Override + public void addSession(WebSocketSession wsSession) { + Object sessionKey = sessionKeyGenerator.sessionKey(wsSession); + Map sessions = this.sessionKeyToWsSessions.get(sessionKey); + if (sessions == null) { + sessions = new ConcurrentHashMap<>(); + this.sessionKeyToWsSessions.putIfAbsent(sessionKey, sessions); + sessions = this.sessionKeyToWsSessions.get(sessionKey); + } + sessions.put(wsSession.getId(), wsSession); + } + + /** + * 删除一个 session + * @param session WebSocketSession + */ + @Override + public void removeSession(WebSocketSession session) { + Object sessionKey = sessionKeyGenerator.sessionKey(session); + String wsSessionId = session.getId(); + + Map sessions = this.sessionKeyToWsSessions.get(sessionKey); + if (sessions != null) { + boolean result = sessions.remove(wsSessionId) != null; + if (log.isDebugEnabled()) { + log.debug("Removal of " + wsSessionId + " was " + result); + } + if (sessions.isEmpty()) { + this.sessionKeyToWsSessions.remove(sessionKey); + if (log.isDebugEnabled()) { + log.debug("Removed the corresponding HTTP Session for " + wsSessionId + + " since it contained no WebSocket mappings"); + } + } + } + } + + /** + * 获取当前所有在线的 session + * @return Collection session集合 + */ + @Override + public Collection getSessions() { + return sessionKeyToWsSessions.values().stream().flatMap(x -> x.values().stream()).collect(Collectors.toList()); + } + + /** + * 根据指定的 sessionKey 获取对应的 wsSessions + * @param sessionKey wsSession 标识 + * @return Collection websocket session集合 + */ + @Override + public Collection getSessions(Object sessionKey) { + return sessionKeyToWsSessions.get(sessionKey).values(); + } + + /** + * 获取所有的 sessionKeys + * @return sessionKey 集合 + */ + @Override + public Collection getSessionKeys() { + return sessionKeyToWsSessions.keySet(); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/MapSessionWebSocketHandlerDecorator.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/MapSessionWebSocketHandlerDecorator.java new file mode 100644 index 000000000..ba7880450 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/MapSessionWebSocketHandlerDecorator.java @@ -0,0 +1,57 @@ +package com.pig4cloud.pigx.common.websocket.session; + +import com.pig4cloud.pigx.common.websocket.handler.ConcurrentWebSocketSessionOptions; +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; + +/** + * WebSocketHandler 装饰器，该装饰器主要用于在开启和关闭连接时，进行session的映射存储与释放 + * + * @author Hccake 2020/12/31 + * @version 1.0 + */ +public class MapSessionWebSocketHandlerDecorator extends WebSocketHandlerDecorator { + + private final WebSocketSessionStore webSocketSessionStore; + + private final ConcurrentWebSocketSessionOptions concurrentWebSocketSessionOptions; + + public MapSessionWebSocketHandlerDecorator(WebSocketHandler delegate, WebSocketSessionStore webSocketSessionStore, + ConcurrentWebSocketSessionOptions concurrentWebSocketSessionOptions) { + super(delegate); + this.webSocketSessionStore = webSocketSessionStore; + this.concurrentWebSocketSessionOptions = concurrentWebSocketSessionOptions; + } + + /** + * websocket 连接时执行的动作 + * @param wsSession websocket session 对象 + * @throws Exception 异常对象 + */ + @Override + public void afterConnectionEstablished(WebSocketSession wsSession) throws Exception { + // 包装一层，防止并发发送出现问题 + if (Boolean.TRUE.equals(concurrentWebSocketSessionOptions.isEnable())) { + wsSession = new ConcurrentWebSocketSessionDecorator(wsSession, + concurrentWebSocketSessionOptions.getSendTimeLimit(), + concurrentWebSocketSessionOptions.getBufferSizeLimit(), + concurrentWebSocketSessionOptions.getOverflowStrategy()); + } + webSocketSessionStore.addSession(wsSession); + } + + /** + * websocket 关闭连接时执行的动作 + * @param wsSession websocket session 对象 + * @param closeStatus 关闭状态对象 + * @throws Exception 异常对象 + */ + @Override + public void afterConnectionClosed(WebSocketSession wsSession, CloseStatus closeStatus) throws Exception { + webSocketSessionStore.removeSession(wsSession); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/SessionKeyGenerator.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/SessionKeyGenerator.java similarity index 87% rename from pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/SessionKeyGenerator.java rename to pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/SessionKeyGenerator.java index bccb5dd3c..d29e7c91e 100644 --- a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/SessionKeyGenerator.java +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/SessionKeyGenerator.java @@ -1,4 +1,4 @@ -package com.pig4cloud.pigx.common.websocket.holder; +package com.pig4cloud.pigx.common.websocket.session; import org.springframework.web.socket.WebSocketSession; diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/WebSocketSessionStore.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/WebSocketSessionStore.java new file mode 100644 index 000000000..3e906c961 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/session/WebSocketSessionStore.java @@ -0,0 +1,43 @@ +package com.pig4cloud.pigx.common.websocket.session; + +import org.springframework.web.socket.WebSocketSession; + +import java.util.Collection; + +/** + * @author hccake + */ +public interface WebSocketSessionStore { + + /** + * 添加一个 session + * @param session 待添加的 WebSocketSession + */ + void addSession(WebSocketSession session); + + /** + * 删除一个 session + * @param session WebSocketSession + */ + void removeSession(WebSocketSession session); + + /** + * 获取当前所有在线的 wsSessions + * @return Collection websocket session集合 + */ + Collection getSessions(); + + /** + * 根据指定的 sessionKey 获取对应的 wsSessions + * @param sessionKey wsSession 标识 + * @return Collection websocket session集合 + */ + Collection getSessions(Object sessionKey); + + /** + * 获取所有的 sessionKeys + * @return sessionKey 集合 + */ + Collection getSessionKeys(); + +} diff --git a/pigx-common/pigx-common-websocket/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/pigx-common/pigx-common-websocket/src/main/resources/META-INF/additional-spring-configuration-metadata.json new file mode 100644 index 000000000..9824783bb --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/resources/META-INF/additional-spring-configuration-metadata.json @@ -0,0 +1,32 @@ +{ + "properties": [ + { + "name": "ballcat.websocket.concurrent.buffer-size-limit", + "type": "java.lang.Integer", + "sourceType": "com.pig4cloud.pigx.common.websocket.handler.ConcurrentWebSocketSessionOptions", + "description": "发送消息缓冲上限 (byte)", + "defaultValue": 102400 + }, + { + "name": "ballcat.websocket.concurrent.enable", + "type": "java.lang.Boolean", + "sourceType": "com.pig4cloud.pigx.common.websocket.handler.ConcurrentWebSocketSessionOptions", + "description": "是否在多线程环境下进行发送", + "defaultValue": false + }, + { + "name": "ballcat.websocket.concurrent.overflow-strategy", + "type": "org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator$OverflowStrategy", + "sourceType": "com.pig4cloud.pigx.common.websocket.handler.ConcurrentWebSocketSessionOptions", + "description": "缓冲区溢出时的执行策略", + "defaultValue": "org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator.OverflowStrategy.TERMINATE" + }, + { + "name": "ballcat.websocket.concurrent.send-time-limit", + "type": "java.lang.Integer", + "sourceType": "com.pig4cloud.pigx.common.websocket.handler.ConcurrentWebSocketSessionOptions", + "description": "发送时间的限制（ms）", + "defaultValue": 5000 + } + ] +} \ No newline at end of file diff --git a/pigx-common/pigx-common-websocket/src/main/resources/META-INF/spring.factories b/pigx-common/pigx-common-websocket/src/main/resources/META-INF/spring.factories index ddaf06a0c..6b43c96c6 100644 --- a/pigx-common/pigx-common-websocket/src/main/resources/META-INF/spring.factories +++ b/pigx-common/pigx-common-websocket/src/main/resources/META-INF/spring.factories @@ -1,4 +1,2 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ - com.pig4cloud.pigx.common.websocket.config.WebSocketAutoConfiguration,\ - com.pig4cloud.pigx.common.websocket.config.RedisMessageDistributorConfiguration,\ - com.pig4cloud.pigx.common.websocket.config.LocalMessageDistributorConfiguration + com.pig4cloud.pigx.common.websocket.config.WebSocketAutoConfiguration \ No newline at end of file