Merge remote-tracking branch 'public/feat/websocket' into leng_dev

# Conflicts:
#	pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java
#	pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java
#	pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java

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<WebSocketSession> 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) {

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);
 	}
 
 }

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 {

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);
 	}
 
 }

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);
+
+}

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<JsonMessageHandler> jsonMessageHandlerList;
-
 	@Bean
 	@ConditionalOnMissingBean
 	public WebSocketConfigurer webSocketConfigurer(List<HandshakeInterceptor> 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<JsonMessageHandler<? extends JsonWebSocketMessage>> jsonMessageHandlerList) {
+		return new JsonMessageHandlerInitializer(jsonMessageHandlerList);
 	}
 
 }

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();
-	}
-
-}
+}

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;
+
+	}
 
 }

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;
 

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<Object> 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<WebSocketSession> 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);
+				}
+			}
+		}
+	}
+
+}

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);
+	}
 
 	/**
 	 * 消息分发

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<Object> 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);
-	}
-
 }

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);
+
 }

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageSender.java

@@ -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<Object> sessionKeys = messageDO.getSessionKeys();
-		if (needBroadcast != null && needBroadcast) {
-			// 广播信息
-			WebSocketMessageSender.broadcast(messageText);
-		}
-		else if (CollectionUtil.isNotEmpty(sessionKeys)) {
-			// 指定用户发送
-			for (Object sessionKey : sessionKeys) {
-				WebSocketMessageSender.send(sessionKey, messageText);
-			}
-		}
-	}
-
-}

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<String> 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);
+		}
 	}
 
 }

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));
+	}
+
+}

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisWebsocketMessageListener.java

@@ -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<String> 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);
-		}
-	}
-
-}

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);
+	}
+
+}

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;
+
+}

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<? extends JsonWebSocketMessage> 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<JsonWebSocketMessage> jsonMessageHandler = JsonMessageHandlerHolder.getHandler(messageType);
+		if (jsonMessageHandler == null) {
+			log.error("[handleTextMessage] 消息类型（{}）不存在对应的消息处理器", messageType);
+			return;
+		}
+		// 消息处理
+		Class<? extends JsonWebSocketMessage> messageClass = jsonMessageHandler.getMessageClass();
+		JsonWebSocketMessage websocketMessageJson;
+		try {
+			websocketMessageJson = MAPPER.treeToValue(jsonNode, messageClass);
+		}
+		catch (JsonProcessingException e) {
+			throw new ErrorJsonMessageException("消息序列化异常，class " + messageClass);
+		}
+		jsonMessageHandler.handle(session, websocketMessageJson);
 	}
 
 }

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;

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<String, JsonMessageHandler> MESSAGE_HANDLER_MAP = new ConcurrentHashMap<>();
+	private static final Map<String, JsonMessageHandler<JsonWebSocketMessage>> MESSAGE_HANDLER_MAP = new ConcurrentHashMap<>();
 
-	public static JsonMessageHandler getHandler(String type) {
+	public static JsonMessageHandler<JsonWebSocketMessage> getHandler(String type) {
 		return MESSAGE_HANDLER_MAP.get(type);
 	}
 
-	public static void addHandler(JsonMessageHandler jsonMessageHandler) {
+	public static void addHandler(JsonMessageHandler<JsonWebSocketMessage> jsonMessageHandler) {
 		MESSAGE_HANDLER_MAP.put(jsonMessageHandler.type(), jsonMessageHandler);
 	}
 

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;
+
+/**
+ * <p>
+ * JsonMessageHandler 初始化器
+ * <p/>
+ * 将所有的 jsonMessageHandler 收集到 JsonMessageHandlerHolder 中
+ *
+ * @author Hccake
+ */
+@RequiredArgsConstructor
+public class JsonMessageHandlerInitializer {
+
+	private final List<JsonMessageHandler<? extends JsonWebSocketMessage>> jsonMessageHandlerList;
+
+	@SuppressWarnings("unchecked")
+	@PostConstruct
+	public void initJsonMessageHandlerHolder() {
+		for (JsonMessageHandler<? extends JsonWebSocketMessage> jsonMessageHandler : jsonMessageHandlerList) {
+			JsonMessageHandlerHolder.addHandler((JsonMessageHandler<JsonWebSocketMessage>) jsonMessageHandler);
+		}
+	}
+
+}

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java

@@ -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);
-	}
-
-}

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/WebSocketSessionHolder.java

@@ -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<Object, WebSocketSession> 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<WebSocketSession> session集合
-	 */
-	public static Collection<WebSocketSession> getSessions() {
-		return USER_SESSION_MAP.values();
-	}
-
-	/**
-	 * 获取所有在线的用户标识
-	 * @return Set<Object> session唯一标识集合
-	 */
-	public static Set<Object> getSessionKeys() {
-		return USER_SESSION_MAP.keySet();
-	}
-
-}

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<Object, Map<String, WebSocketSession>> 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<String, WebSocketSession> 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<String, WebSocketSession> 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<WebSocketSession> session集合
+	 */
+	@Override
+	public Collection<WebSocketSession> getSessions() {
+		return sessionKeyToWsSessions.values().stream().flatMap(x -> x.values().stream()).collect(Collectors.toList());
+	}
+
+	/**
+	 * 根据指定的 sessionKey 获取对应的 wsSessions
+	 * @param sessionKey wsSession 标识
+	 * @return Collection<WebSocketSession> websocket session集合
+	 */
+	@Override
+	public Collection<WebSocketSession> getSessions(Object sessionKey) {
+		return sessionKeyToWsSessions.get(sessionKey).values();
+	}
+
+	/**
+	 * 获取所有的 sessionKeys
+	 * @return sessionKey 集合
+	 */
+	@Override
+	public Collection<Object> getSessionKeys() {
+		return sessionKeyToWsSessions.keySet();
+	}
+
+}

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);
+	}
+
+}

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;
 

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<WebSocketSession> websocket session集合
+	 */
+	Collection<WebSocketSession> getSessions();
+
+	/**
+	 * 根据指定的 sessionKey 获取对应的 wsSessions
+	 * @param sessionKey wsSession 标识
+	 * @return Collection<WebSocketSession> websocket session集合
+	 */
+	Collection<WebSocketSession> getSessions(Object sessionKey);
+
+	/**
+	 * 获取所有的 sessionKeys
+	 * @return sessionKey 集合
+	 */
+	Collection<Object> getSessionKeys();
+
+}

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
+    }
+  ]
+}

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