修改启动异常问题

puzhibing

2026-05-14 cb352bea5670dcd79d4267ebabf7616f7aa61d6f

 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/handler/WebSocketTextFrameHandler.java

@@ -2,6 +2,8 @@

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiaopeng.api.vo.automaticPilot.AppUserPosition;
import com.xiaopeng.common.redis.service.RedisService;
import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
@@ -9,12 +11,19 @@
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
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.geo.GeoJsonPoint;
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 javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.UUID;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * TextWebSocketFrame 表示一个文本桢
@@ -29,20 +38,33 @@

    @Autowired
    private WebSocketUserConnectionManager userConnectionManager;
    

    @Autowired
    private WebSocketAuthHandler authHandler;

    @Resource
    private MongoTemplate mongoTemplate;

    Map<String, BusinessMsgHandler> businessMsgHandlers;

    public WebSocketTextFrameHandler(List<BusinessMsgHandler> handlers) {
        this.businessMsgHandlers = handlers.stream()
                .collect(Collectors.toMap(
                        BusinessMsgHandler::getMessageType,
                        Function.identity()
                ));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        String message = msg.text();
        log.info("[服务器] : 收到消息 -> {}", message);
        

        try {
            // 尝试解析消息，检查是否包含用户认证信息
            JSONObject messageObj = JSON.parseObject(message);
            String messageType = messageObj.getString("type");
            

            if ("auth".equals(messageType)) {
                // 处理用户认证消息
                handleUserAuth(ctx, messageObj);
@@ -54,12 +76,13 @@
                handleBusinessMessage(ctx, messageObj);
            }
        } catch (Exception e) {
            e.printStackTrace();
            // 如果不是JSON格式，按普通文本处理
            log.debug("消息不是JSON格式，按普通文本处理: {}", message);
            ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间: " + LocalDateTime.now() + "->" + message));
        }
    }
    

    /**
     * 处理用户认证消息
     */
@@ -69,18 +92,18 @@
            WebSocketAuthHandler.UserAuthInfo authInfo = authHandler.extractAuthInfo(messageObj);
            String appUserId = authInfo.getAppUserId();
            String token = authInfo.getToken();
            

            if (!StringUtils.hasText(appUserId)) {
                sendAuthFailureResponse(ctx, "用户ID不能为空");
                return;
            }
            

            // 进行用户认证
            WebSocketAuthHandler.AuthResult authResult = authHandler.authenticateUser(appUserId, token);
            if (authResult.isSuccess()) {
                // 认证成功，将连接与用户关联
                userConnectionManager.addUserConnection(appUserId, ctx.channel());
                

                // 发送认证成功响应
                JSONObject response = new JSONObject();
                response.put("type", "auth_success");
@@ -91,14 +114,14 @@
//                response.put("nickName", authResult.getAppUser().getNickName());
                response.put("message", "认证成功");
                response.put("timestamp", LocalDateTime.now().toString());
                

                ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
                log.info("用户 {} ({}) 认证成功，ChannelId: {}", 
                log.info("用户 {} ({}) 认证成功，ChannelId: {}",
                        appUserId, uuid, ctx.channel().id().asLongText());
            } else {
                // 认证失败
                sendAuthFailureResponse(ctx, authResult.getMessage());
                log.warn("用户 {} 认证失败: {}, ChannelId: {}", 
                log.warn("用户 {} 认证失败: {}, ChannelId: {}",
                        appUserId, authResult.getMessage(), ctx.channel().id().asLongText());
            }
        } catch (Exception e) {
@@ -106,7 +129,7 @@
            sendAuthFailureResponse(ctx, "认证过程中发生错误: " + e.getMessage());
        }
    }
    

    /**
     * 发送认证失败响应
     */
@@ -115,51 +138,53 @@
        response.put("type", "auth_failed");
        response.put("message", message);
        response.put("timestamp", LocalDateTime.now().toString());
        

        ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
    }
    

    /**
     * 处理心跳消息
     */
    private void handlePing(ChannelHandlerContext ctx, JSONObject messageObj) {
        String appUserId = userConnectionManager.getUserIdByChannelId(ctx.channel().id().asLongText());
        

        JSONObject response = new JSONObject();
        response.put("type", "pong");
        log.info("心跳用户id：{}",appUserId);
        response.put("appUserId", appUserId);
        response.put("timestamp", LocalDateTime.now().toString());
        

        ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
    }
    

    /**
     * 处理业务消息
     */
    private void handleBusinessMessage(ChannelHandlerContext ctx, JSONObject messageObj) {
        String appUserId = userConnectionManager.getUserIdByChannelId(ctx.channel().id().asLongText());
        

        if (appUserId != null) {
            log.info("收到用户 {} 的业务消息: {}", appUserId, messageObj.toJSONString());
            

            businessMsgHandlers.get(messageObj.getString("type")).handle(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());
            
            ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
//            JSONObject response = new JSONObject();
//            response.put("type", "message_response");
//            response.put("appUserId", appUserId);
//            response.put("originalMessage", messageObj);
//            response.put("serverTime", LocalDateTime.now().toString());
//
//            ctx.channel().writeAndFlush(new TextWebSocketFrame(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());
            

            ctx.channel().writeAndFlush(new TextWebSocketFrame(response.toJSONString()));
        }
    }
@@ -180,7 +205,7 @@
        log.info("handlerRemoved 被调用: {}", ctx.channel().id().asLongText());
        // shortText 可能会重复
        log.info("handlerRemoved 被调用: {}", ctx.channel().id().asShortText());
        

        // 从用户连接管理器中移除连接
        if (userConnectionManager != null) {
            userConnectionManager.removeUserConnection(ctx.channel());
@@ -192,4 +217,5 @@
        log.error("WebSocket连接异常: {}", cause.getMessage(), cause);
        ctx.channel().close();
    }

}