提交 2e53164f authored 作者: 吕本才's avatar 吕本才

feat(config): 并发发送ping,增加日志打印查看问题

上级 b5f97239
...@@ -33,7 +33,7 @@ public class WebSocketConfig implements WebSocketConfigurer { ...@@ -33,7 +33,7 @@ public class WebSocketConfig implements WebSocketConfigurer {
} }
@Bean @Bean
public WebSocketHandler webSocketHandler() { public TemporaryActivityTaskClockSocketHandler webSocketHandler() {
return new TemporaryActivityTaskClockSocketHandler(); return new TemporaryActivityTaskClockSocketHandler();
} }
......
...@@ -7,20 +7,20 @@ import lombok.extern.slf4j.Slf4j; ...@@ -7,20 +7,20 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler; import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.io.IOException; import java.io.IOException;
import java.util.Map; import java.util.Map;
import java.util.concurrent.*; import java.util.concurrent.*;
@Slf4j @Slf4j
@Component //@Component
public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandler { public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandler {
// 存储在线会话(用户ID -> 会话) // 存储在线会话(用户ID -> 会话)
private static final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>(); private static final ConcurrentHashMap<String, WebSocketSession> sessions = new ConcurrentHashMap<>();
// 心跳检测线程池 // 心跳检测线程池
private final ScheduledExecutorService heartBeatExecutor = new ScheduledThreadPoolExecutor( private final ScheduledExecutorService heartBeatExecutor = new ScheduledThreadPoolExecutor(
1, 1,
...@@ -35,6 +35,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle ...@@ -35,6 +35,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle
private final ObjectMapper objectMapper = new ObjectMapper(); private final ObjectMapper objectMapper = new ObjectMapper();
public TemporaryActivityTaskClockSocketHandler() { public TemporaryActivityTaskClockSocketHandler() {
log.info("TemporaryActivityTaskClockSocketHandler 初始化");
// 初始化心跳检测(每30秒发送一次ping) // 初始化心跳检测(每30秒发送一次ping)
heartBeatExecutor.scheduleAtFixedRate(this::sendHeartBeat, 30, 30, TimeUnit.SECONDS); heartBeatExecutor.scheduleAtFixedRate(this::sendHeartBeat, 30, 30, TimeUnit.SECONDS);
} }
...@@ -75,7 +76,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle ...@@ -75,7 +76,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle
log.error("用户[" + userId + "]消息格式错误:" + payload); log.error("用户[" + userId + "]消息格式错误:" + payload);
sendToUser(userId, "用户[" + userId + "]消息格式错误:" ); sendToUser(userId, "用户[" + userId + "]消息格式错误:" );
} }
handleMessageType(messageBean); // handleMessageType(messageBean);
} }
} }
private void handleMessageType(MessageBean messageBean) { private void handleMessageType(MessageBean messageBean) {
...@@ -103,7 +104,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle ...@@ -103,7 +104,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle
/** /**
* 发送心跳(ping) * 发送心跳(ping)
*/ */
private void sendHeartBeat() { private void sendHeartBeat() {
for (Map.Entry<String, WebSocketSession> entry : sessions.entrySet()) { for (Map.Entry<String, WebSocketSession> entry : sessions.entrySet()) {
WebSocketSession session = entry.getValue(); WebSocketSession session = entry.getValue();
String userId = entry.getKey(); String userId = entry.getKey();
...@@ -115,6 +116,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle ...@@ -115,6 +116,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle
try { try {
session.close(); session.close();
} catch (IOException ex) { } catch (IOException ex) {
log.error("关闭会话失败", ex);
ex.printStackTrace(); ex.printStackTrace();
} }
sessions.remove(userId); sessions.remove(userId);
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论