From fd6875763b8dfef10f9ea96941e58fa51caee5a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=B7=E5=86=B7?= <2270033969@qq.com> Date: Tue, 3 Sep 2024 11:30:50 +0800 Subject: [PATCH] =?UTF-8?q?refactor(websocket):=20=E6=9B=B4=E6=8D=A2?= =?UTF-8?q?=E6=88=90ConcurrentWebSocketSessionDecorator=E9=9D=9E=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=E5=BC=8F=E7=9A=84sendMessage=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/WebSocketHandlerConfig.java | 4 +- .../websocket/config/WebSocketProperties.java | 63 +++++++++++-------- .../MapSessionWebSocketHandlerDecorator.java | 63 +++++++++++-------- 3 files changed, 75 insertions(+), 55 deletions(-) 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..099113c8f 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 @@ -51,7 +51,7 @@ public class WebSocketHandlerConfig { public WebSocketHandler webSocketHandler1(@Autowired(required = false) SessionKeyGenerator sessionKeyGenerator) { CustomWebSocketHandler customWebSocketHandler = new CustomWebSocketHandler(); if (webSocketProperties.isMapSession()) { - return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator); + return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator, webSocketProperties); } return customWebSocketHandler; } @@ -63,7 +63,7 @@ public class WebSocketHandlerConfig { PlanTextMessageHandler planTextMessageHandler) { CustomWebSocketHandler customWebSocketHandler = new CustomWebSocketHandler(planTextMessageHandler); if (webSocketProperties.isMapSession()) { - return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator); + return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator,webSocketProperties); } return customWebSocketHandler; } 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..3c78793d2 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 @@ -12,38 +12,47 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(WebSocketProperties.PREFIX) public class WebSocketProperties { - public static final String PREFIX = "pigx.websocket"; + public static final String PREFIX = "pigx.websocket"; - /** - * 路径: 无参: /ws 有参: PathVariable: 单参: /ws/{test} 多参: /ws/{test1}/{test2} query: - * /ws?uid=1&name=test - * - */ - private String path = "/ws/info"; + /** + * 路径: 无参: /ws 有参: PathVariable: 单参: /ws/{test} 多参: /ws/{test1}/{test2} query: + * /ws?uid=1&name=test + */ + private String path = "/ws/info"; - /** - * 允许访问源 - */ - private String allowOrigins = "*"; + /** + * 允许访问源 + */ + private String allowOrigins = "*"; - /** - * 是否支持部分消息 - */ - private boolean supportPartialMessages = false; + /** + * 是否支持部分消息 + */ + private boolean supportPartialMessages = false; - /** - * 心跳处理 - */ - private boolean heartbeat = true; + /** + * 心跳处理 + */ + private boolean heartbeat = true; - /** - * 是否开启对session的映射记录 - */ - private boolean mapSession = true; + /** + * 是否开启对session的映射记录 + */ + private boolean mapSession = true; - /** - * 消息分发器：local | redis，默认 local, 如果自定义的话，可以配置为其他任意值 - */ - private String messageDistributor = MessageDistributorTypeConstants.LOCAL; + /** + * 消息分发器：local | redis，默认 local, 如果自定义的话，可以配置为其他任意值 + */ + private String messageDistributor = MessageDistributorTypeConstants.LOCAL; + + /** + * 配置的发送时间限制（毫秒）。 + */ + private Integer sendTimeLimit = 10000; + + /** + * 配置的缓冲区大小限制 + */ + private Integer sendBufferSizeLimit = 64000; } 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 index 74c3ac43d..b7b4729dc 100644 --- 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 @@ -1,8 +1,10 @@ package com.pig4cloud.pigx.common.websocket.holder; +import com.pig4cloud.pigx.common.websocket.config.WebSocketProperties; 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; /** @@ -13,34 +15,43 @@ import org.springframework.web.socket.handler.WebSocketHandlerDecorator; */ public class MapSessionWebSocketHandlerDecorator extends WebSocketHandlerDecorator { - private final SessionKeyGenerator sessionKeyGenerator; + private final SessionKeyGenerator sessionKeyGenerator; - public MapSessionWebSocketHandlerDecorator(WebSocketHandler delegate, SessionKeyGenerator sessionKeyGenerator) { - super(delegate); - this.sessionKeyGenerator = sessionKeyGenerator; - } + private final WebSocketProperties webSocketProperties; - /** - * 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); - } + public MapSessionWebSocketHandlerDecorator(WebSocketHandler delegate, SessionKeyGenerator sessionKeyGenerator, WebSocketProperties webSocketProperties) { + super(delegate); + this.sessionKeyGenerator = sessionKeyGenerator; + this.webSocketProperties = webSocketProperties; + } - /** - * 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); - } + /** + * websocket 连接时执行的动作 + * + * @param session websocket session 对象 + * ConcurrentWebSocketSessionDecorator确保一次只有一个线程可以发送消息。 + * 如果某个线程的发送操作较慢，其他线程尝试发送消息时将无法获取锁，消息将被缓冲。 + * 此时，装饰器会检查指定的缓冲区大小限制和发送时间限制，如果超过这些限制，会关闭会话 + * @throws Exception 异常对象 + */ + @Override + public void afterConnectionEstablished(final WebSocketSession session) throws Exception { + Object sessionKey = sessionKeyGenerator.sessionKey(session); + WebSocketSessionHolder.addSession(sessionKey + , new ConcurrentWebSocketSessionDecorator(session, webSocketProperties.getSendTimeLimit(), webSocketProperties.getSendBufferSizeLimit())); + } + + /** + * 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); + } }