package com.xiaopeng.pushMessage.webSocket.handler; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import com.xiaopeng.api.dto.account.XpOperationsUserPositionDTO; import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; 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 javax.annotation.PostConstruct; import java.time.LocalDateTime; import java.util.UUID; /** * Spring WebSocket 消息处理器 * 路径: /websocket,共用 Spring Boot HTTP 端口 * 替代原 Netty 独立端口方案 */ @Component @Slf4j public class PushMessageWebSocketHandler extends TextWebSocketHandler { @Autowired private WebSocketUserConnectionManager userConnectionManager; @Autowired private WebSocketAuthHandler authHandler; private ObjectMapper om; @PostConstruct void setupObjectMapper() { // 允许对象忽略json中不存在的属性 om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); om.registerModule(new JavaTimeModule()); om = new ObjectMapper(); } // ==================== 连接生命周期 ==================== @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { log.info("WebSocket 连接建立,sessionId: {}", session.getId()); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { log.info("WebSocket 连接关闭,sessionId: {}, status: {}", session.getId(), status); userConnectionManager.removeUserConnection(session); } @Override public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { log.error("WebSocket 传输异常,sessionId: {}, error: {}", session.getId(), exception.getMessage(), exception); if (session.isOpen()) { session.close(CloseStatus.SERVER_ERROR); } } // ==================== 消息处理 ==================== @Override protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { String payload = message.getPayload(); log.info("[服务器] 收到消息 -> {}", payload); try { if (payload.startsWith("pos:")) { if (log.isDebugEnabled()) log.debug("收到运维人员位置上报消息:{}",payload); XpOperationsUserPositionDTO xpOperationsUserPositionDTO = om.readValue(payload.substring(4), XpOperationsUserPositionDTO.class); saveToMongo(xpOperationsUserPositionDTO); } JSONObject messageObj = JSON.parseObject(payload); String messageType = messageObj.getString("type"); if ("auth".equals(messageType)) { handleUserAuth(session, messageObj); } else if ("ping".equals(messageType)) { handlePing(session); } else { handleBusinessMessage(session, messageObj); } } catch (Exception e) { log.debug("消息非 JSON 格式,按普通文本处理: {}", payload); sendText(session, "服务器时间: " + LocalDateTime.now() + "->" + payload); } } @Autowired private MongoTemplate mongoTemplate; private void saveToMongo(XpOperationsUserPositionDTO dto) { XpOperationsUserPositionDTO vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("id").is(dto.getId())), XpOperationsUserPositionDTO.class); if (null == vehiclePosition) { mongoTemplate.save(dto); }else{ Update update = new Update() .set("id",dto.getId()) .set("latitude", dto.getLatitude()) .set("latitude", dto.getLongitude()); mongoTemplate.updateFirst(new Query(Criteria.where("id").is(dto.getId())), update, XpOperationsUserPositionDTO.class); } } // ==================== 业务处理 ==================== /** * 处理用户认证消息 */ private void handleUserAuth(WebSocketSession session, JSONObject messageObj) { try { WebSocketAuthHandler.UserAuthInfo authInfo = authHandler.extractAuthInfo(messageObj); String appUserId = authInfo.getAppUserId(); String token = authInfo.getToken(); if (!StringUtils.hasText(appUserId)) { sendAuthFailure(session, "用户ID不能为空"); return; } WebSocketAuthHandler.AuthResult authResult = authHandler.authenticateUser(appUserId, token); if (authResult.isSuccess()) { userConnectionManager.addUserConnection(appUserId, session); // TODO String uuid = UUID.randomUUID().toString(); JSONObject response = new JSONObject(); response.put("type", "auth_success"); response.put("appUserId", appUserId); response.put("nickName", uuid); response.put("message", "认证成功"); response.put("timestamp", LocalDateTime.now().toString()); sendText(session, response.toJSONString()); log.info("用户 {} ({}) 认证成功,sessionId: {}", appUserId, uuid, session.getId()); } else { sendAuthFailure(session, authResult.getMessage()); log.warn("用户 {} 认证失败: {}, sessionId: {}", appUserId, authResult.getMessage(), session.getId()); } } catch (Exception e) { log.error("处理用户认证时发生错误", e); sendAuthFailure(session, "认证过程中发生错误: " + e.getMessage()); } } private void sendAuthFailure(WebSocketSession session, String message) { JSONObject response = new JSONObject(); response.put("type", "auth_failed"); response.put("message", message); response.put("timestamp", LocalDateTime.now().toString()); sendText(session, response.toJSONString()); } /** * 处理心跳消息 */ private void handlePing(WebSocketSession session) { String appUserId = userConnectionManager.getUserIdBySessionId(session.getId()); JSONObject response = new JSONObject(); response.put("type", "pong"); response.put("appUserId", appUserId); response.put("timestamp", LocalDateTime.now().toString()); log.info("心跳用户id:{}", appUserId); sendText(session, response.toJSONString()); } /** * 处理业务消息 */ private void handleBusinessMessage(WebSocketSession session, JSONObject messageObj) { String appUserId = userConnectionManager.getUserIdBySessionId(session.getId()); if (appUserId != null) { log.info("收到用户 {} 的业务消息: {}", appUserId, messageObj.toJSONString()); JSONObject response = new JSONObject(); response.put("type", "message_response"); response.put("appUserId", appUserId); response.put("originalMessage", messageObj); response.put("serverTime", LocalDateTime.now().toString()); sendText(session, response.toJSONString()); } else { log.warn("收到未认证连接的业务消息: {}", messageObj.toJSONString()); JSONObject response = new JSONObject(); response.put("type", "auth_required"); response.put("message", "请先进行用户认证"); response.put("timestamp", LocalDateTime.now().toString()); sendText(session, response.toJSONString()); } } // ==================== 工具方法 ==================== /** * 发送文本消息(线程安全) */ private void sendText(WebSocketSession session, String text) { try { if (session.isOpen()) { synchronized (session) { session.sendMessage(new TextMessage(text)); } } } catch (Exception e) { log.error("发送消息失败,sessionId: {}, error: {}", session.getId(), e.getMessage()); } } }