sx/map
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

websocket并发发送消息

Browse Source
This commit is contained in:
谷成伟 2024-10-22 12:40:22 +08:00
parent 406d7fca24
commit 5af50a92df
1 changed files with 5 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
das/src/main/java/com/das/modules/node/handler/NodeMessageHandler.java
View File

@ -15,6 +15,7 @@ import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.web.socket.*; import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
import org.springframework.web.socket.handler.TextWebSocketHandler; import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.util.concurrent.*; import java.util.concurrent.*;
@ -24,7 +25,7 @@ import java.util.concurrent.*;
public class NodeMessageHandler extends TextWebSocketHandler { public class NodeMessageHandler extends TextWebSocketHandler {
public static final Long HEARTBEAT_TIMEOUT = 1000 * 60 * 1L; public static final Long HEARTBEAT_TIMEOUT = 1000 * 60 * 1L;
private ConcurrentHashMap<Long, WebSocketSession> onlineSessions = new ConcurrentHashMap<>(16); private ConcurrentHashMap<Long, ConcurrentWebSocketSessionDecorator> onlineSessions = new ConcurrentHashMap<>(16);
private DataService dataService; private DataService dataService;
@Override @Override
@ -43,7 +44,7 @@ public class NodeMessageHandler extends TextWebSocketHandler {
} }
else { else {
log.info("IP: {} 准许连接, NodeId:{}, Version: {}, sessionId: {}", remoteIp, nodeId, version, session.getId()); log.info("IP: {} 准许连接, NodeId:{}, Version: {}, sessionId: {}", remoteIp, nodeId, version, session.getId());
onlineSessions.put(nodeId, session); onlineSessions.put(nodeId, new ConcurrentWebSocketSessionDecorator(session, 5 * 1000, 2 * 1024 * 1024));
} }
} }
@ -108,7 +109,7 @@ public class NodeMessageHandler extends TextWebSocketHandler {
*/ */
@Scheduled(cron = "0/15 * * * * ?") @Scheduled(cron = "0/15 * * * * ?")
public void sendHeartbeat(){ public void sendHeartbeat(){
for (WebSocketSession session : onlineSessions.values()) { for (ConcurrentWebSocketSessionDecorator session : onlineSessions.values()) {
//判断心跳是否超时，超时则主动断开连接 //判断心跳是否超时，超时则主动断开连接
Long lastPongTime = (Long) session.getAttributes().get(NodeConstant.LAST_PONG_TIME); Long lastPongTime = (Long) session.getAttributes().get(NodeConstant.LAST_PONG_TIME);
if (lastPongTime != null && ((System.currentTimeMillis() - lastPongTime) > HEARTBEAT_TIMEOUT)){ if (lastPongTime != null && ((System.currentTimeMillis() - lastPongTime) > HEARTBEAT_TIMEOUT)){
@ -144,7 +145,7 @@ public class NodeMessageHandler extends TextWebSocketHandler {
* @param nodeId * @param nodeId
*/ */
public void sendActionMessage(Long nodeId, TerminalMessage message){ public void sendActionMessage(Long nodeId, TerminalMessage message){
WebSocketSession session = onlineSessions.get(nodeId); ConcurrentWebSocketSessionDecorator session = onlineSessions.get(nodeId);
if (session != null){ if (session != null){
try { try {
session.sendMessage(new TextMessage(message.toJsonString())); session.sendMessage(new TextMessage(message.toJsonString()));