zianxin/pigx
1
0 Fork 0
mirror of http://git.pig4cloud.com/pig/pigx.git synced 2026-11-08 04:39:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(websocket): 更换成ConcurrentWebSocketSessionDecorator非阻塞式的sendMessage发送

Browse Source
This commit is contained in:
冷冷 committed 2024-09-03 11:31:17 +08:00
1 parent 2071195f7d
commit fd6875763b
3 files changed
+75 -55
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java
+2 -2
View File
@@ -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;
}
pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java
+36 -27
View File
@@ -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;
}
pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java
+37 -26
View File
@@ -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);
}
}