aidlab/aida_back
7
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ws 引入心跳

Browse Source
This commit is contained in:
xupei
2024-11-11 16:34:55 +08:00
parent 8da51a0a82
commit 3cf0570912
1 changed files with 21 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
src/main/java/com/ai/da/common/websocket/NotificationConnection.java
View File

@@ -17,38 +17,29 @@ import java.util.concurrent.ConcurrentHashMap;
public class NotificationConnection { public class NotificationConnection {
//连接超时 //连接超时
public static final long MAX_TIME_OUT = 2 * 60 * 1000; public static final long MAX_TIME_OUT = 3 * 60 * 1000;
static Map<String,Session> sessionMap = new ConcurrentHashMap<>();
private Session session; private Session session;
private Long userId;
// 这里用ConcurrentHashMap 因为他是一个线程安全的Map // 这里用ConcurrentHashMap 因为他是一个线程安全的Map
private static ConcurrentHashMap<NotificationConnection, Long> websockets = new ConcurrentHashMap<>(); private static ConcurrentHashMap<Long, NotificationConnection> websockets = new ConcurrentHashMap<>();
//连接建立时执行的操作
/*@OnOpen
public void onOpen(Session session){
sessionMap.put(session.getId(),session);
log.info("websocket is open, sessionId: {}",session.getId());
}*/
@OnOpen @OnOpen
public void onOpen(Session session, @PathParam("id") String id) { // 接收到前端传来的用户ID public void onOpen(Session session, @PathParam("id") String id) { // 接收到前端传来的用户ID
this.session = session; this.session = session;
// this.session.setMaxIdleTimeout(MAX_TIME_OUT); this.userId = Long.valueOf(id);
websockets.put(this, Long.parseLong(id)); //将ID作为key，当前的对象作为Value this.session.setMaxIdleTimeout(MAX_TIME_OUT);
websockets.put(Long.parseLong(id), this); //将ID作为key，当前的对象作为Value
log.info("【建立连接】 用户为：{}", this.session); log.info("【建立连接】 用户为：{}", this.session);
log.info("【建立连接】 用户Id为：{}", id); log.info("【建立连接】 用户Id为：{}", id);
log.info("【建立连接】 总数为：{}", websockets.size()); log.info("【当前连接总数】 为：{}", websockets.size());
} }
@OnClose @OnClose
public void onClose(CloseReason reason) { public void onClose(CloseReason reason) {
if (websockets.containsKey(this)) { log.info("【连接断开】 用户为：{}, sessionId: {}, 原因为{}", this.userId, this.session.getId(), reason);
websockets.remove(this); // 将当前的对象从集合中删除 log.info("【当前连接总数】 为：{}", websockets.size());
log.info("【连接断开】 用户为：{}, sessionId: {}, 原因为{}", websockets.get(this), this.session.getId(), reason); websockets.remove(this.userId); // 将当前的对象从集合中删除
}
// log.info("【连接断开】 总数为：{}", websockets.size());
} }
/** /**
@@ -57,42 +48,29 @@ public class NotificationConnection {
*/ */
@OnError @OnError
public void onError(Throwable throwable) { public void onError(Throwable throwable) {
log.info("【连接异常】 用户为：{} , sessionId: {}", websockets.get(this), this.session.getId(), throwable); log.info("【连接异常】 用户为：{} , sessionId: {}", this.userId, this.session.getId(), throwable);
websockets.remove(this); // 将当前的对象从集合中删除 websockets.remove(this.userId); // 将当前的对象从集合中删除
log.info("【当前连接总数】 为：{}", websockets.size());
} }
//收到了客户端消息执行的操作 //收到了客户端消息执行的操作
@OnMessage @OnMessage
public void onMessage(Session session, String text){ public void onMessage(String text){
log.info("收到了一条消息："+text); log.info("收到了一条来自 {} 的消息：{}", this.userId, text);
// return "已收到你的消息"; // return "已收到你的消息";
if (text.equals("PING")){ if (text.equals("PING")){
sendMsg("PONG", websockets.get(this)); sendMsg("PONG", this.userId);
session.setMaxIdleTimeout(MAX_TIME_OUT);
} }
} }
/*//连接关闭的时候执行的操作
@OnClose
public void onClose(Session session){
sessionMap.remove(session.getId());
log.info("websocket is close, sessionId: {}",session.getId());
}
public void sendMsg(String message) throws IOException {
for(String key:sessionMap.keySet()){
sessionMap.get(key).getBasicRemote().sendText(message);
}
}*/
public void sendMsg(String message, Long userId) { public void sendMsg(String message, Long userId) {
if (userId == null) { // 如果等于null则证明是群发 if (userId == null) { // 如果等于null则证明是群发
// 获取当前Map的一个迭代器，遍历Map的方式有很多种，看着来 // 获取当前Map的一个迭代器，遍历Map的方式有很多种，看着来
// 这个就是遍历这个集合的过程.... // 这个就是遍历这个集合的过程....
for (Map.Entry<NotificationConnection, Long> entry : websockets.entrySet()) { for (Map.Entry<Long, NotificationConnection> entry : websockets.entrySet()) {
// 获取每一个Entry实例 // 获取每一个Entry实例
// 获取每一个Value，而这个Value就是WebSocket的实例 // 获取每一个Value，而这个Value就是WebSocket的实例
NotificationConnection webSocket = entry.getKey(); NotificationConnection webSocket = entry.getValue();
// 接下来就是遍历群发 // 接下来就是遍历群发
log.info("广播消息 【给用户】 ：{}发送消息【{}】", webSocket, message); log.info("广播消息 【给用户】 ：{}发送消息【{}】", webSocket, message);
try { try {
@@ -104,12 +82,12 @@ public class NotificationConnection {
} else { // 如果不是群发，则判断ID，其余步骤一致 } else { // 如果不是群发，则判断ID，其余步骤一致
// 获取当前Map的一个迭代器，遍历Map的方式有很多种，看着来 // 获取当前Map的一个迭代器，遍历Map的方式有很多种，看着来
// 这个就是遍历这个集合的过程.... // 这个就是遍历这个集合的过程....
for (Map.Entry<NotificationConnection, Long> entry : websockets.entrySet()) { for (Map.Entry<Long, NotificationConnection> entry : websockets.entrySet()) {
// 获取每一个Entry实例 // 获取每一个Entry实例
// 获取每一个Value，而这个Value就是WebSocket的实例 // 获取每一个Value，而这个Value就是WebSocket的实例
NotificationConnection webSocket = entry.getKey(); NotificationConnection webSocket = entry.getValue();
// 获取每一个Key，这个Key就是用户ID // 获取每一个Key，这个Key就是用户ID
Long key = entry.getValue(); Long key = entry.getKey();
// 判断用户ID与当前的Key相等 // 判断用户ID与当前的Key相等
if (userId.equals(key)) { if (userId.equals(key)) {
log.info("私发消息 【给用户】 ：{}发送消息【{}】", key, message); // 打印 log.info("私发消息 【给用户】 ：{}发送消息【{}】", key, message); // 打印