fix:websocket共用http端口

zhuml4

2026-03-04 9c19c883f2088324965c6d50e5514691db5508fe

fix:websocket共用http端口

 xp-robotaxi-itinerary-boot/src/main/java/com/xiaopeng/itinerary/service/impl/XpPricingServiceImpl.java

@@ -60,6 +60,7 @@
 * @author liu
 * @since 2026-02-03
 */
@Slf4j
@Service
@Slf4j
public class XpPricingServiceImpl extends ServiceImpl<XpPricingMapper, XpPricing> implements XpPricingService {

 xp-robotaxi-push-message-boot/pom.xml

@@ -125,6 +125,12 @@
            <version>2.0.5</version>
        </dependency>

        <!-- Spring WebSocket 支持（共用 Spring Boot HTTP 端口） -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>

    </dependencies>

    <build>

 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/config/WebSocketConfig.java

New file
@@ -0,0 +1,27 @@
package com.xiaopeng.pushMessage.webSocket.config;

import com.xiaopeng.pushMessage.webSocket.handler.PushMessageWebSocketHandler;
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;

/**
 * Spring WebSocket 配置
 * 注册 /websocket 端点，共用 Spring Boot HTTP 端口
 */
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Autowired
    private PushMessageWebSocketHandler pushMessageWebSocketHandler;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(pushMessageWebSocketHandler, "/websocket")
                // 允许所有跨域来源，生产环境可按需收紧
                .setAllowedOrigins("*");
    }
}

 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/config/WebSocketServer.java

File was deleted

 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/config/WebSocketServerConfig.java

File was deleted

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

New file
@@ -0,0 +1,188 @@
package com.xiaopeng.pushMessage.webSocket.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
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 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;

    // ==================== 连接生命周期 ====================

    @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 {
            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);
        }
    }

    // ==================== 业务处理 ====================

    /**
     * 处理用户认证消息
     */
    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());
        }
    }
}

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

@@ -1,195 +1,195 @@
package com.xiaopeng.pushMessage.webSocket.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.time.LocalDateTime;
import java.util.Objects;
import java.util.UUID;

/**
 * TextWebSocketFrame 表示一个文本桢
 * 使用@Sharable注解使其可以被多个连接共享
 * 支持基于AppUser的连接管理
 */
@Component
@Slf4j
@ChannelHandler.Sharable
public class WebSocketTextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {


    @Autowired
    private WebSocketUserConnectionManager userConnectionManager;
    
    @Autowired
    private WebSocketAuthHandler authHandler;

    @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);
            } else if ("ping".equals(messageType)) {
                // 处理心跳消息
                handlePing(ctx, messageObj);
            } else {
                // 处理普通业务消息
                handleBusinessMessage(ctx, messageObj);
            }
        } catch (Exception e) {
            // 如果不是JSON格式，按普通文本处理
            log.debug("消息不是JSON格式，按普通文本处理: {}", message);
            ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间: " + LocalDateTime.now() + "->" + message));
        }
    }
    
    /**
     * 处理用户认证消息
     */
    private void handleUserAuth(ChannelHandlerContext ctx, JSONObject messageObj) {
        try {
            // 提取认证信息
            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");
                response.put("appUserId", appUserId);
                // TODO
                String uuid = UUID.randomUUID().toString();
                response.put("nickName", uuid);
//                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: {}", 
                        appUserId, uuid, ctx.channel().id().asLongText());
            } else {
                // 认证失败
                sendAuthFailureResponse(ctx, authResult.getMessage());
                log.warn("用户 {} 认证失败: {}, ChannelId: {}", 
                        appUserId, authResult.getMessage(), ctx.channel().id().asLongText());
            }
        } catch (Exception e) {
            log.error("处理用户认证时发生错误", e);
            sendAuthFailureResponse(ctx, "认证过程中发生错误: " + e.getMessage());
        }
    }
    
    /**
     * 发送认证失败响应
     */
    private void sendAuthFailureResponse(ChannelHandlerContext ctx, String message) {
        JSONObject response = new JSONObject();
        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());
            
            // 回复消息
            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()));
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // id 表示唯一的值 LongText是唯一的
        log.info("handlerAdded 被调用: {}", ctx.channel().id().asLongText());
        // shortText 可能会重复
        log.info("handlerAdded 被调用: {}", ctx.channel().id().asShortText());

        // 注意：用户连接管理在认证时添加，这里不添加
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        // id 表示唯一的值 LongText是唯一的
        log.info("handlerRemoved 被调用: {}", ctx.channel().id().asLongText());
        // shortText 可能会重复
        log.info("handlerRemoved 被调用: {}", ctx.channel().id().asShortText());
        
        // 从用户连接管理器中移除连接
        if (userConnectionManager != null) {
            userConnectionManager.removeUserConnection(ctx.channel());
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("WebSocket连接异常: {}", cause.getMessage(), cause);
        ctx.channel().close();
    }
}
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.xiaopeng.pushMessage.webSocket.service.WebSocketUserConnectionManager;
//import io.netty.channel.ChannelHandler;
//import io.netty.channel.ChannelHandlerContext;
//import io.netty.channel.SimpleChannelInboundHandler;
//import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//import org.springframework.util.StringUtils;
//
//import java.time.LocalDateTime;
//import java.util.Objects;
//import java.util.UUID;
//
///**
// * TextWebSocketFrame 表示一个文本桢
// * 使用@Sharable注解使其可以被多个连接共享
// * 支持基于AppUser的连接管理
// */
//@Component
//@Slf4j
//@ChannelHandler.Sharable
//public class WebSocketTextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
//
//
//    @Autowired
//    private WebSocketUserConnectionManager userConnectionManager;
//
//    @Autowired
//    private WebSocketAuthHandler authHandler;
//
//    @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);
//            } else if ("ping".equals(messageType)) {
//                // 处理心跳消息
//                handlePing(ctx, messageObj);
//            } else {
//                // 处理普通业务消息
//                handleBusinessMessage(ctx, messageObj);
//            }
//        } catch (Exception e) {
//            // 如果不是JSON格式，按普通文本处理
//            log.debug("消息不是JSON格式，按普通文本处理: {}", message);
//            ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间: " + LocalDateTime.now() + "->" + message));
//        }
//    }
//
//    /**
//     * 处理用户认证消息
//     */
//    private void handleUserAuth(ChannelHandlerContext ctx, JSONObject messageObj) {
//        try {
//            // 提取认证信息
//            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");
//                response.put("appUserId", appUserId);
//                // TODO
//                String uuid = UUID.randomUUID().toString();
//                response.put("nickName", uuid);
////                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: {}",
//                        appUserId, uuid, ctx.channel().id().asLongText());
//            } else {
//                // 认证失败
//                sendAuthFailureResponse(ctx, authResult.getMessage());
//                log.warn("用户 {} 认证失败: {}, ChannelId: {}",
//                        appUserId, authResult.getMessage(), ctx.channel().id().asLongText());
//            }
//        } catch (Exception e) {
//            log.error("处理用户认证时发生错误", e);
//            sendAuthFailureResponse(ctx, "认证过程中发生错误: " + e.getMessage());
//        }
//    }
//
//    /**
//     * 发送认证失败响应
//     */
//    private void sendAuthFailureResponse(ChannelHandlerContext ctx, String message) {
//        JSONObject response = new JSONObject();
//        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());
//
//            // 回复消息
//            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()));
//        }
//    }
//
//    @Override
//    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
//        // id 表示唯一的值 LongText是唯一的
//        log.info("handlerAdded 被调用: {}", ctx.channel().id().asLongText());
//        // shortText 可能会重复
//        log.info("handlerAdded 被调用: {}", ctx.channel().id().asShortText());
//
//        // 注意：用户连接管理在认证时添加，这里不添加
//    }
//
//    @Override
//    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
//        // id 表示唯一的值 LongText是唯一的
//        log.info("handlerRemoved 被调用: {}", ctx.channel().id().asLongText());
//        // shortText 可能会重复
//        log.info("handlerRemoved 被调用: {}", ctx.channel().id().asShortText());
//
//        // 从用户连接管理器中移除连接
//        if (userConnectionManager != null) {
//            userConnectionManager.removeUserConnection(ctx.channel());
//        }
//    }
//
//    @Override
//    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
//        log.error("WebSocket连接异常: {}", cause.getMessage(), cause);
//        ctx.channel().close();
//    }
//}

 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/webSocket/service/WebSocketUserConnectionManager.java

@@ -1,175 +1,155 @@
package com.xiaopeng.pushMessage.webSocket.service;

import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于AppUser的WebSocket连接管理器
 * 支持按用户ID管理连接，一个用户可以有多个设备连接
 * 基于 AppUser 的 WebSocket 连接管理器
 * 使用 Spring WebSocketSession（共用 Spring Boot HTTP 端口）
 * 支持按用户 ID 管理连接，一个用户可以有多个设备连接
 */
@Component
@Slf4j
public class WebSocketUserConnectionManager {
    

    /**
     * 存储用户的所有连接
     * Key: 用户ID (appUserId)
     * Value: 该用户的所有连接Channel列表
     * Value: 该用户的所有 WebSocketSession 集合
     */
    private static final ConcurrentHashMap<String, Set<Channel>> userConnections = new ConcurrentHashMap<>();
    
    private static final ConcurrentHashMap<String, Set<WebSocketSession>> userConnections = new ConcurrentHashMap<>();

    /**
     * 存储连接与用户的映射关系
     * Key: ChannelId.asLongText()
     * Key: session.getId()
     * Value: 用户ID (appUserId)
     */
    private static final ConcurrentHashMap<String, String> channelToUser = new ConcurrentHashMap<>();
    
    private static final ConcurrentHashMap<String, String> sessionToUser = new ConcurrentHashMap<>();

    /**
     * 存储所有连接
     * Key: ChannelId.asLongText()
     * Value: Channel
     * Key: session.getId()
     * Value: WebSocketSession
     */
    private static final ConcurrentHashMap<String, Channel> allConnections = new ConcurrentHashMap<>();
    
    private static final ConcurrentHashMap<String, WebSocketSession> allConnections = new ConcurrentHashMap<>();

    // ==================== 连接管理 ====================

    /**
     * 添加用户连接
     * @param appUserId 用户ID
     * @param channel WebSocket连接通道
     */
    public void addUserConnection(String appUserId, Channel channel) {
        String channelId = channel.id().asLongText();
        
        // 添加到用户连接映射
        userConnections.computeIfAbsent(appUserId, k -> ConcurrentHashMap.newKeySet())
                .add(channel);
        
        // 添加到连接用户映射
        channelToUser.put(channelId, appUserId);
        
        // 添加到所有连接
        allConnections.put(channelId, channel);
        
        log.info("用户连接已添加，用户ID: {}, ChannelId: {}, 该用户连接数: {}", 
                appUserId, channelId, userConnections.get(appUserId).size());
    public void addUserConnection(String appUserId, WebSocketSession session) {
        String sessionId = session.getId();

        userConnections.computeIfAbsent(appUserId, k -> ConcurrentHashMap.newKeySet()).add(session);
        sessionToUser.put(sessionId, appUserId);
        allConnections.put(sessionId, session);

        log.info("用户连接已添加，用户ID: {}, sessionId: {}, 该用户连接数: {}",
                appUserId, sessionId, userConnections.get(appUserId).size());
    }
    

    /**
     * 移除用户连接
     * @param channel WebSocket连接通道
     */
    public void removeUserConnection(Channel channel) {
        String channelId = channel.id().asLongText();
        String appUserId = channelToUser.get(channelId);
        
    public void removeUserConnection(WebSocketSession session) {
        String sessionId = session.getId();
        String appUserId = sessionToUser.get(sessionId);

        if (appUserId != null) {
            // 从用户连接中移除
            Set<Channel> userChannels = userConnections.get(appUserId);
            if (userChannels != null) {
                userChannels.remove(channel);
                // 如果用户没有连接了，删除用户映射
                if (userChannels.isEmpty()) {
            Set<WebSocketSession> userSessions = userConnections.get(appUserId);
            if (userSessions != null) {
                userSessions.remove(session);
                if (userSessions.isEmpty()) {
                    userConnections.remove(appUserId);
                }
            }
            
            // 从连接用户映射中移除
            channelToUser.remove(channelId);
            
            log.info("用户连接已移除，用户ID: {}, ChannelId: {}, 该用户剩余连接数: {}", 
                    appUserId, channelId, userChannels != null ? userChannels.size() : 0);
            sessionToUser.remove(sessionId);
            log.info("用户连接已移除，用户ID: {}, sessionId: {}, 该用户剩余连接数: {}",
                    appUserId, sessionId, userSessions != null ? userSessions.size() : 0);
        }
        
        // 从所有连接中移除
        allConnections.remove(channelId);

        allConnections.remove(sessionId);
    }
    

    // ==================== 消息发送 ====================

    /**
     * 向指定用户的所有设备发送消息
     * @param appUserId 用户ID
     * @param message 消息内容
     * @return 成功发送的设备数
     */
    public int sendMessageToUser(String appUserId, String message) {
        Set<Channel> userChannels = userConnections.get(appUserId);
        if (userChannels == null || userChannels.isEmpty()) {
        Set<WebSocketSession> sessions = userConnections.get(appUserId);
        if (sessions == null || sessions.isEmpty()) {
            log.warn("用户 {} 没有活跃连接", appUserId);
            return 0;
        }
        

        int successCount = 0;
        for (Channel channel : userChannels) {
            if (channel != null && channel.isActive()) {
        for (WebSocketSession session : sessions) {
            if (session != null && session.isOpen()) {
                try {
                    channel.writeAndFlush(new TextWebSocketFrame(message));
                    synchronized (session) {
                        session.sendMessage(new TextMessage(message));
                    }
                    successCount++;
                } catch (Exception e) {
                    log.error("向用户 {} 发送消息失败，连接ID: {}, 错误: {}", 
                            appUserId, channel.id().asLongText(), e.getMessage());
                    log.error("向用户 {} 发送消息失败，sessionId: {}, 错误: {}",
                            appUserId, session.getId(), e.getMessage());
                }
            }
        }
        
        log.info("向用户 {} 发送消息完成，成功发送到 {} 个设备，消息内容: {}", 
                appUserId, successCount, message);

        log.info("向用户 {} 发送消息完成，成功发送到 {} 个设备", appUserId, successCount);
        return successCount;
    }
    

    /**
     * 向指定连接发送消息
     * @param channelId 连接ID
     * @param message 消息内容
     * @return 是否发送成功
     * 向指定 session 发送消息
     */
    public boolean sendMessageToChannel(String channelId, String message) {
        Channel channel = allConnections.get(channelId);
        if (channel != null && channel.isActive()) {
    public boolean sendMessageToChannel(String sessionId, String message) {
        WebSocketSession session = allConnections.get(sessionId);
        if (session != null && session.isOpen()) {
            try {
                channel.writeAndFlush(new TextWebSocketFrame(message));
                log.info("消息已发送到连接: {}, 内容: {}", channelId, message);
                synchronized (session) {
                    session.sendMessage(new TextMessage(message));
                }
                log.info("消息已发送到连接: {}", sessionId);
                return true;
            } catch (Exception e) {
                log.error("发送消息失败，连接ID: {}, 错误: {}", channelId, e.getMessage());
                log.error("发送消息失败，sessionId: {}, 错误: {}", sessionId, e.getMessage());
                return false;
            }
        } else {
            log.warn("连接不存在或已关闭，ChannelId: {}", channelId);
            log.warn("连接不存在或已关闭，sessionId: {}", sessionId);
            return false;
        }
    }
    

    /**
     * 向所有连接广播消息
     * @param message 消息内容
     * @return 成功发送的连接数
     */
    public int broadcastMessage(String message) {
        int successCount = 0;
        for (Channel channel : allConnections.values()) {
            if (channel != null && channel.isActive()) {
        for (WebSocketSession session : allConnections.values()) {
            if (session != null && session.isOpen()) {
                try {
                    channel.writeAndFlush(new TextWebSocketFrame(message));
                    synchronized (session) {
                        session.sendMessage(new TextMessage(message));
                    }
                    successCount++;
                } catch (Exception e) {
                    log.error("广播消息失败，连接ID: {}, 错误: {}", 
                            channel.id().asLongText(), e.getMessage());
                    log.error("广播消息失败，sessionId: {}, 错误: {}", session.getId(), e.getMessage());
                }
            }
        }
        log.info("广播消息完成，成功发送到 {} 个连接，消息内容: {}", successCount, message);
        log.info("广播消息完成，成功发送到 {} 个连接", successCount);
        return successCount;
    }
    

    /**
     * 向多个用户发送消息
     * @param appUserIds 用户ID列表
     * @param message 消息内容
     * @return 成功发送的用户数
     */
    public int sendMessageToUsers(List<String> appUserIds, String message) {
        int successUserCount = 0;
@@ -182,100 +162,62 @@
        log.info("向 {} 个用户发送消息完成，成功发送到 {} 个用户", appUserIds.size(), successUserCount);
        return successUserCount;
    }
    
    /**
     * 获取用户的所有连接
     * @param appUserId 用户ID
     * @return 连接列表
     */
    public Set<Channel> getUserConnections(String appUserId) {

    // ==================== 查询 ====================

    public Set<WebSocketSession> getUserConnections(String appUserId) {
        return userConnections.getOrDefault(appUserId, Collections.emptySet());
    }
    
    /**
     * 获取用户连接数
     * @param appUserId 用户ID
     * @return 连接数
     */

    public int getUserConnectionCount(String appUserId) {
        Set<Channel> userChannels = userConnections.get(appUserId);
        return userChannels != null ? userChannels.size() : 0;
        Set<WebSocketSession> sessions = userConnections.get(appUserId);
        return sessions != null ? sessions.size() : 0;
    }
    
    /**
     * 获取总连接数
     * @return 总连接数
     */

    public int getTotalConnectionCount() {
        return allConnections.size();
    }
    
    /**
     * 获取在线用户数
     * @return 在线用户数
     */

    public int getOnlineUserCount() {
        return userConnections.size();
    }
    
    /**
     * 获取所有在线用户ID
     * @return 用户ID集合
     */

    public Set<String> getOnlineUserIds() {
        return userConnections.keySet();
    }
    
    /**
     * 检查用户是否在线
     * @param appUserId 用户ID
     * @return 是否在线
     */

    public boolean isUserOnline(String appUserId) {
        Set<Channel> userChannels = userConnections.get(appUserId);
        if (userChannels == null || userChannels.isEmpty()) {
        Set<WebSocketSession> sessions = userConnections.get(appUserId);
        if (sessions == null || sessions.isEmpty()) {
            return false;
        }
        // 检查是否有活跃连接
        return userChannels.stream().anyMatch(Channel::isActive);
        return sessions.stream().anyMatch(WebSocketSession::isOpen);
    }
    
    /**
     * 检查连接是否存在
     * @param channelId 连接ID
     * @return 是否存在
     */
    public boolean isConnectionExists(String channelId) {
        Channel channel = allConnections.get(channelId);
        return channel != null && channel.isActive();

    public boolean isConnectionExists(String sessionId) {
        WebSocketSession session = allConnections.get(sessionId);
        return session != null && session.isOpen();
    }
    

    /**
     * 根据连接ID获取用户ID
     * @param channelId 连接ID
     * @return 用户ID
     * 根据 sessionId 获取用户ID
     */
    public String getUserIdByChannelId(String channelId) {
        return channelToUser.get(channelId);
    public String getUserIdBySessionId(String sessionId) {
        return sessionToUser.get(sessionId);
    }
    
    /**
     * 获取连接统计信息
     * @return 统计信息
     */

    public Map<String, Object> getConnectionStats() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("totalConnections", getTotalConnectionCount());
        stats.put("onlineUsers", getOnlineUserCount());
        stats.put("onlineUserIds", getOnlineUserIds());
        
        // 统计每个用户的连接数

        Map<String, Integer> userConnectionCounts = new HashMap<>();
        for (Map.Entry<String, Set<Channel>> entry : userConnections.entrySet()) {
        for (Map.Entry<String, Set<WebSocketSession>> entry : userConnections.entrySet()) {
            userConnectionCounts.put(entry.getKey(), entry.getValue().size());
        }
        stats.put("userConnectionCounts", userConnectionCounts);
        

        return stats;
    }
}

 xp-robotaxi-push-message-boot/src/main/resources/application-local.properties

File was deleted

 xp-robotaxi-push-message-boot/src/main/resources/bootstrap.yml

@@ -1,7 +1,7 @@
server:
  port: 6090
  tomcat:
    basedir: /data/tomcat
    basedir: data/tomcat
# Spring
spring:
  application: