xiaohuo
8 months ago
parent
0950044a88
commit
7aa07df754
@ -1,20 +1,27 @@ |
||||
package com.alops.system.config; |
||||
package com.alops.framework.config; |
||||
|
||||
import com.alops.system.hander.WebSocketServer; |
||||
|
||||
import com.alops.framework.hander.WebSocketServer; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.context.annotation.Configuration; |
||||
import org.springframework.web.socket.config.annotation.EnableWebSocket; |
||||
import org.springframework.web.socket.config.annotation.WebSocketConfigurer; |
||||
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; |
||||
|
||||
//理解为WebSocket的controller
|
||||
//注册端点，允许跨域
|
||||
@Configuration |
||||
@EnableWebSocket |
||||
public class WebSocketConfig implements WebSocketConfigurer { |
||||
|
||||
@Autowired |
||||
private WebSocketServer webSocketServer; |
||||
|
||||
@Autowired |
||||
private WebSocketHandshakeInterceptor handshakeInterceptor; |
||||
|
||||
@Override |
||||
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { |
||||
|
||||
registry.addHandler(new WebSocketServer(), "/ws/onlineMessage") |
||||
registry.addHandler(webSocketServer, "/ws/onlineMessage") |
||||
.addInterceptors(handshakeInterceptor) |
||||
.setAllowedOrigins("*"); |
||||
|
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,65 @@ |
||||
package com.alops.framework.config; |
||||
|
||||
import com.alops.common.core.domain.model.LoginUser; |
||||
import com.alops.framework.web.service.TokenService; |
||||
import lombok.extern.slf4j.Slf4j; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.http.server.ServerHttpRequest; |
||||
import org.springframework.http.server.ServerHttpResponse; |
||||
|
||||
import org.springframework.stereotype.Component; |
||||
import org.springframework.web.socket.WebSocketHandler; |
||||
import org.springframework.web.socket.server.HandshakeInterceptor; |
||||
|
||||
import java.util.List; |
||||
import java.util.Map; |
||||
//拦截器设置，首先在springscrity放行
|
||||
@Slf4j |
||||
@Component |
||||
public class WebSocketHandshakeInterceptor implements HandshakeInterceptor { |
||||
|
||||
@Autowired |
||||
private TokenService tokenService; |
||||
|
||||
@Override |
||||
public boolean beforeHandshake(ServerHttpRequest request, |
||||
ServerHttpResponse response, |
||||
WebSocketHandler wsHandler, |
||||
Map<String, Object> attributes) { |
||||
|
||||
try { |
||||
List<String> protocols = request.getHeaders().get("Sec-WebSocket-Protocol"); |
||||
if (protocols == null || protocols.isEmpty()) { |
||||
log.warn("【WS握手失败】未携带Token"); |
||||
return false; |
||||
} |
||||
|
||||
String token = protocols.get(0); |
||||
LoginUser loginUser = tokenService.getLoginUser(token); |
||||
if (loginUser == null) { |
||||
log.warn("【WS握手失败】无效Token"); |
||||
return false; |
||||
} |
||||
|
||||
// 验证并刷新 Token
|
||||
tokenService.verifyToken(loginUser); |
||||
|
||||
// 设置返回协议头（必须，否则浏览器报错）
|
||||
response.getHeaders().add("Sec-WebSocket-Protocol", token); |
||||
|
||||
// 把用户信息写入 session attributes
|
||||
attributes.put("loginUser", loginUser); |
||||
log.info("【WS握手成功】用户：{}", loginUser.getUsername()); |
||||
return true; |
||||
|
||||
} catch (Exception e) { |
||||
log.error("【WS握手异常】", e); |
||||
return false; |
||||
} |
||||
} |
||||
|
||||
@Override |
||||
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, |
||||
WebSocketHandler wsHandler, Exception exception) { |
||||
} |
||||
} |
||||
@ -0,0 +1,43 @@ |
||||
package com.alops.framework.hander; |
||||
|
||||
import com.alops.common.core.domain.model.LoginUser; |
||||
|
||||
import com.alops.system.service.impl.WebSocketService; |
||||
import org.slf4j.Logger; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.stereotype.Component; |
||||
import org.springframework.web.socket.CloseStatus; |
||||
import org.springframework.web.socket.WebSocketSession; |
||||
import org.springframework.web.socket.handler.TextWebSocketHandler; |
||||
import org.slf4j.LoggerFactory; |
||||
@Component |
||||
public class WebSocketServer extends TextWebSocketHandler { |
||||
|
||||
@Autowired |
||||
private WebSocketService webSocketService; |
||||
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(WebSocketServer.class); |
||||
@Override |
||||
public void afterConnectionEstablished(WebSocketSession session) throws Exception { |
||||
LoginUser loginUser = (LoginUser) session.getAttributes().get("loginUser"); |
||||
if (loginUser != null && loginUser.getUser() != null) { |
||||
Long userId = loginUser.getUser().getUserId(); |
||||
webSocketService.addOnlineUser(userId, session); |
||||
logger.info("WebSocket 连接成功，用户上线: {}", userId); |
||||
} else { |
||||
logger.warn("无效连接，关闭 session: {}", session.getId()); |
||||
session.close(); |
||||
} |
||||
} |
||||
|
||||
@Override |
||||
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { |
||||
LoginUser loginUser = (LoginUser) session.getAttributes().get("loginUser"); |
||||
if (loginUser != null && loginUser.getUser() != null) { |
||||
Long userId = loginUser.getUser().getUserId(); |
||||
webSocketService.removeOnlineUser(userId); |
||||
logger.info("WebSocket 连接关闭，用户下线: {}, 状态: {}", userId, status); |
||||
} |
||||
} |
||||
} |
||||
@ -1,153 +0,0 @@ |
||||
package com.alops.system.hander; |
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.stereotype.Component; |
||||
import org.springframework.util.MultiValueMap; |
||||
import org.springframework.web.socket.CloseStatus; |
||||
import org.springframework.web.socket.TextMessage; |
||||
import org.springframework.web.socket.WebSocketSession; |
||||
import org.springframework.web.socket.handler.TextWebSocketHandler; |
||||
import org.springframework.web.util.UriComponentsBuilder; |
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
|
||||
import java.util.Date; |
||||
import java.util.HashMap; |
||||
import java.util.Map; |
||||
import java.util.Set; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
|
||||
@Component |
||||
public class WebSocketServer extends TextWebSocketHandler { |
||||
private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class); |
||||
// 存储在线用户 (Key: 用户ID, Value: WebSocketSession)
|
||||
public static final ConcurrentHashMap<Long, WebSocketSession> onlineUsers = new ConcurrentHashMap<>(); |
||||
|
||||
|
||||
private static ObjectMapper objectMapper; |
||||
|
||||
|
||||
@Autowired |
||||
public void setObjectMapper(ObjectMapper mapper) { |
||||
WebSocketServer.objectMapper = mapper; |
||||
} |
||||
|
||||
@Override |
||||
public void afterConnectionEstablished(WebSocketSession session) throws Exception { |
||||
Long userId = getUserIdFromSession(session); |
||||
if (userId != null) { |
||||
logger.info("WebSocket 连接成功，用户上线: " + userId); |
||||
onlineUsers.put(userId, session); |
||||
} else { |
||||
logger.warn("无效的用户连接，关闭连接: " + session.getId()); |
||||
session.close(); // 没有 userId 的连接直接关闭
|
||||
} |
||||
} |
||||
|
||||
@Override |
||||
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { |
||||
Long userId = getUserIdFromSession(session); |
||||
if (userId != null) { |
||||
|
||||
logger.info("WebSocket 连接关闭，用户下线: " + userId + ", 关闭状态: " + status); |
||||
onlineUsers.remove(userId); |
||||
|
||||
} |
||||
} |
||||
|
||||
// 获取 WebSocketSession 中的 userId，并转换为 Long 类型
|
||||
public static Long getUserIdFromSession(WebSocketSession session) { |
||||
String query = session.getUri().getQuery(); |
||||
logger.info("WebSocket连接查询参数: {}", query); |
||||
|
||||
if (query != null) { |
||||
MultiValueMap<String, String> params = UriComponentsBuilder.newInstance() |
||||
.query(query) |
||||
.build() |
||||
.getQueryParams(); |
||||
String userIdStr = params.getFirst("userId"); |
||||
if (userIdStr != null) { |
||||
try { |
||||
return Long.parseLong(userIdStr); |
||||
} catch (NumberFormatException e) { |
||||
logger.error("WebSocket连接无效的用户ID: {}", userIdStr, e); |
||||
} |
||||
} else { |
||||
logger.warn("WebSocket连接没有 userId 参数"); |
||||
} |
||||
} else { |
||||
logger.warn("WebSocket连接没有查询参数"); |
||||
} |
||||
return null; |
||||
} |
||||
//
|
||||
// // 发送消息给指定用户
|
||||
// public void sendMessageToUser(Long userId, String content, String title) throws Exception {
|
||||
// // 获取对应用户的 WebSocket session
|
||||
// WebSocketSession session = onlineUsers.get(userId);
|
||||
//
|
||||
// // 检查 session 是否有效且连接未关闭
|
||||
// if (session != null && session.isOpen()) {
|
||||
// // 创建一个 Map 来构建消息内容
|
||||
// Map<String, Object> messageMap = new HashMap<>();
|
||||
// messageMap.put("from UserName", fromUserName); // 发送者姓名
|
||||
// messageMap.put("content", content); // 消息内容
|
||||
//
|
||||
// // 使用 Jackson ObjectMapper 将 Map 转换为 JSON 字符串
|
||||
// ObjectMapper objectMapper = new ObjectMapper();
|
||||
// String jsonMessage = objectMapper.writeValueAsString(messageMap);
|
||||
//
|
||||
// // 发送 JSON 格式的消息
|
||||
// session.sendMessage(new TextMessage(jsonMessage));
|
||||
// }
|
||||
// }
|
||||
|
||||
|
||||
/** |
||||
* 发送通知到指定用户 |
||||
* @param messageId 数据库中站内信的ID |
||||
* @param title 通知标题，例如 "设备异常预警" |
||||
* @param content 通知内容 |
||||
*/ |
||||
public static void sendAlertMessage(Long userId, Long messageId, String title, String content) { |
||||
WebSocketSession session = onlineUsers.get(userId); |
||||
try { |
||||
if (session != null && session.isOpen()) { |
||||
|
||||
// 构建消息内容
|
||||
Map<String, Object> messageMap = new HashMap<>(); |
||||
messageMap.put("messageId", messageId); // 对应数据库ID
|
||||
messageMap.put("title", title); // 弹窗标题
|
||||
messageMap.put("content", content); // 弹窗内容
|
||||
messageMap.put("timestamp", new Date()); // 发送时间
|
||||
|
||||
// 转换为 JSON
|
||||
String jsonMessage = objectMapper.writeValueAsString(messageMap); |
||||
|
||||
// 发送
|
||||
session.sendMessage(new TextMessage(jsonMessage)); |
||||
logger.info("成功向用户 [{}] 发送通知: {}", session.getId(), jsonMessage); |
||||
} |
||||
} catch (Exception e) { |
||||
logger.error("发送通知失败: messageId=" + messageId, e); |
||||
} |
||||
} |
||||
|
||||
|
||||
public void sendMessageToUser1(Long userId, String message) throws Exception { |
||||
WebSocketSession session = onlineUsers.get(userId); |
||||
if (session != null && session.isOpen()) { |
||||
session.sendMessage(new TextMessage(message)); |
||||
} |
||||
} |
||||
|
||||
// 获取所有在线用户 ID (Long 类型)
|
||||
public Set<Long> getOnlineUserIds() { |
||||
return onlineUsers.keySet(); |
||||
} |
||||
|
||||
|
||||
|
||||
} |
||||
@ -0,0 +1,63 @@ |
||||
package com.alops.system.service.impl; |
||||
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
import org.slf4j.LoggerFactory; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.stereotype.Service; |
||||
import org.springframework.web.socket.TextMessage; |
||||
import org.springframework.web.socket.WebSocketSession; |
||||
|
||||
import java.util.Date; |
||||
import java.util.HashMap; |
||||
import java.util.Map; |
||||
import java.util.Set; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
//负责在线用户
|
||||
@Service |
||||
public class WebSocketService { |
||||
|
||||
private final Map<Long, WebSocketSession> onlineUsers = new ConcurrentHashMap<>(); |
||||
|
||||
@Autowired |
||||
private ObjectMapper objectMapper; |
||||
|
||||
// 添加在线用户
|
||||
public void addOnlineUser(Long userId, WebSocketSession session) { |
||||
onlineUsers.put(userId, session); |
||||
} |
||||
|
||||
// 移除在线用户
|
||||
public void removeOnlineUser(Long userId) { |
||||
onlineUsers.remove(userId); |
||||
} |
||||
|
||||
// 获取在线用户
|
||||
public Set<Long> getOnlineUserIds() { |
||||
return onlineUsers.keySet(); |
||||
} |
||||
|
||||
// 给指定用户发送通知
|
||||
public void sendAlertMessage(Long userId, Long messageId, String title, String content) { |
||||
WebSocketSession session = onlineUsers.get(userId); |
||||
try { |
||||
if (session != null && session.isOpen()) { |
||||
Map<String, Object> messageMap = new HashMap<>(); |
||||
messageMap.put("messageId", messageId); |
||||
messageMap.put("title", title); |
||||
messageMap.put("content", content); |
||||
messageMap.put("timestamp", new Date()); |
||||
|
||||
String jsonMessage = objectMapper.writeValueAsString(messageMap); |
||||
session.sendMessage(new TextMessage(jsonMessage)); |
||||
|
||||
LoggerFactory.getLogger(WebSocketService.class) |
||||
.info("成功向用户 [{}] 发送通知: {}", userId, jsonMessage); |
||||
} |
||||
} catch (Exception e) { |
||||
LoggerFactory.getLogger(WebSocketService.class) |
||||
.error("发送通知失败: messageId=" + messageId, e); |
||||
} |
||||
} |
||||
} |
||||
|
||||
Loading…
Reference in new issue