Merge branch 'master' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi

liujie

2026-03-09 65cc684c4c1f4a2d028d4ed5b92994f17492401e

Merge branch 'master' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi

 xp-robotaxi-account-boot/src/main/java/com/xiaopeng/account/controller/SysUserController.java

@@ -321,8 +321,7 @@
        }
        userService.checkUserAllowed(user);
        // userService.checkUserDataScope(user.getId());
        user.setPassword(SecurityUtils.encryptPassword(user.getPassword()));
        return toAjax(userService.updateById(user));
        return toAjax(userService.resetUserPwd(user.getId(), SecurityUtils.encryptPassword(dto.getNewPassword().trim())));
    }

    /**

 xp-robotaxi-account-boot/src/main/java/com/xiaopeng/account/service/impl/SysUserServiceImpl.java

@@ -262,6 +262,6 @@

    @Override
    public boolean resetUserPwd(Long id, String newPassword) {
        return lambdaUpdate().set(SysUser::getId, id).set(SysUser::getPassword, newPassword).update();
        return lambdaUpdate().eq(SysUser::getId, id).set(SysUser::getPassword, newPassword).update();
    }
}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/factory/vehicleConvoy/XpFleetClientFallbackFactory.java

@@ -36,6 +36,12 @@
                // 降级返回空列表
                return R.fail("通过运营人员id查询车队信息失败，"+cause.getMessage());
            }

            @Override
            public R<Long> getUseCountByLocationId(Long locationId) {
                log.error("查询打卡地被使用次数失败，原因: {}", cause.getMessage());
                return R.fail("查询打卡地被使用次数失败");
            }
        };
    }
}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/feignClient/vehicleConvoy/XpFleetClient.java

@@ -38,4 +38,13 @@
     */
    @GetMapping("/inner/v1/fleet/getFleetByOperationsUserId")
    R<XpFleet> getFleetByOperationsUserId(@RequestParam Long operationsUserId);

    /**
     * 根据打卡地址id查询被车辆使用次数
     * @param locationId 打卡地址id
     * @return
     */
    @GetMapping("/inner/fleet/getUseCountByLocationId")
    R<Long> getUseCountByLocationId(@RequestParam Long locationId);

}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/model/operation/XpElectronicFenceArea.java

@@ -15,6 +15,7 @@
import lombok.EqualsAndHashCode;
import lombok.experimental.Accessors;

import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotEmpty;

/**
@@ -57,4 +58,7 @@
    @ApiModelProperty(value = "围栏点位  经度_维度,经度_维度")
    private String fencePoints;

    @ApiModelProperty(value = "运营策略")
    private String operationStrategy;

}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/query/operation/ElectronicPageListQuery.java

@@ -5,6 +5,8 @@
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;

import java.util.List;

@Data
@ApiModel(value = "电子围栏列表查询query")
public class ElectronicPageListQuery extends BasePage {
@@ -13,7 +15,7 @@
    @ApiModelProperty(value = "围栏构成 区域id")
    private String electronicAreaId;
    @ApiModelProperty(value = "围栏状态 0=草稿 1=待启用 2=启用中 3=已停用")
    private Integer status;
    private List<Integer> statusList;
    @ApiModelProperty(value = "创建人")
    private String createBy;
    @ApiModelProperty(value = "创建开始时间")

 xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/OrderMissionController.java

@@ -168,6 +168,11 @@
            redisTemplate.opsForValue().set("CREATE_ORDER:" + dto.getOrderId().toString(), jsonObject.toJSONString(),30, TimeUnit.MINUTES);
            return R.ok(true);
        }
        //先将车辆修改状态，防止在异步通知之前被其他订单选中
        if (!dto.getOrderId().contains("XU") && mqttPublishDto.getCode()== 200) {
            // 修改车子为接单状态
            xpVehicleClient.updateVehicleStatus(new UpdateVehicleStateDto(dto.getVin(), 1));
        }
        return R.ok(false);
    }


 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/controller/XpCheckLocationController.java

@@ -1,5 +1,7 @@
package com.xiaopeng.maintenance.controller;

import com.xiaopeng.api.feignClient.vehicleConvoy.XpFleetClient;
import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
import com.xiaopeng.api.model.maintenance.XpCheckLocation;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.enums.DisabledEnum;
@@ -26,6 +28,7 @@
@RequiredArgsConstructor(onConstructor_ = @Lazy)
public class XpCheckLocationController {
    private final IXpCheckLocationService xpCheckLocationService;
    private final XpFleetClient xpFleetClient;
    private final TokenService tokenService;

    @GetMapping("/api/v1/check-location/list")
@@ -71,6 +74,11 @@
    @DeleteMapping("/api/v1/check-location/deleteById")
    @ApiOperation(value = "删除打卡地", tags = {"管理后台-运维执行"})
    public  R<String> deleteById(@RequestParam Long id){
        // 判断打卡地是否已被使用
        Long data = xpFleetClient.getUseCountByLocationId(id).getData();
        if (data > 0) {
            return R.fail("打卡地已被使用，请先解除关联");
        }
        xpCheckLocationService.removeById(id);
        return R.ok();
    }

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/controller/XpVehicleOperationsController.java

New file
@@ -0,0 +1,163 @@
package com.xiaopeng.maintenance.controller;

import cn.hutool.core.util.IdUtil;
import com.xiaopeng.api.dto.automaticPilot.PushOrderDto;
import com.xiaopeng.api.dto.operation.NewCruiseOrderDto;
import com.xiaopeng.api.feignClient.automaticPilot.OrderMissionClient;
import com.xiaopeng.api.feignClient.operation.XpCruisePlanClient;
import com.xiaopeng.api.feignClient.order.XpCruiseOrderClient;
import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
import com.xiaopeng.api.model.order.XpCruiseOrder;
import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
import com.xiaopeng.api.vo.OrderType;
import com.xiaopeng.api.vo.automaticPilot.VehiclePosition;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.web.page.PageInfo;
import com.xiaopeng.maintenance.dto.VehicleDispatchAcceptDTO;
import com.xiaopeng.maintenance.dto.VehicleDispatchDepartureDTO;
import com.xiaopeng.maintenance.model.query.OperationRecordsQuery;
import com.xiaopeng.maintenance.model.vo.OperationRecordsVO;
import com.xiaopeng.maintenance.service.IXpVehiclePreparationService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
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.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * <p>
 * 车辆管理 前端控制器
 * </p>
 *
 * @author mitao
 * @since 2026-02-03
 */
@Slf4j
@RestController
@Api(tags = "车辆运维")
public class XpVehicleOperationsController {

    @Resource
    private IXpVehiclePreparationService xpVehiclePreparationService;
    @Resource
    private XpVehicleClient xpVehicleClient;
    @Resource
    private XpCruiseOrderClient xpCruiseOrderClient;
    @Resource
    private MongoTemplate mongoTemplate;
    @Resource
    private OrderMissionClient orderMissionClient;
    @Autowired
    private XpCruisePlanClient xpCruisePlanClient;

    /**
     * 车辆详情-车辆调度-发车 收车
     */
    @ApiOperation(value = "车辆运维-车辆调度-发车/收车", tags = {"运维APP-车辆管理"})
    @PostMapping("/api/v1/vehicle/vehicleDispatchDeparture")
    public R<String> vehicleDispatchDeparture(@RequestBody VehicleDispatchDepartureDTO dto) {

        // 判断是否有正在进行的正常订单
        // 查询车辆
        XpVehicle xpVehicle = xpVehicleClient.queryById(dto.getVehicleId()).getData();
        if(xpVehicle.getState()==1){
            return R.fail("车辆正在接单中");
        }
        xpVehicle.setOperatingStatus(dto.getDepartureStatus());
        xpVehicleClient.updateVehicleById(xpVehicle);

        // 发车
        if(dto.getDepartureStatus()==1){
            // 找车辆的任务  生成巡游单子 下发命令
            R<XpCruiseOrder> newCruiseOrder = xpCruisePlanClient.getNewCruiseOrder(new NewCruiseOrderDto(xpVehicle.getId(),xpVehicle.getVehicleVin(),xpVehicle.getFleetId()));
            if(newCruiseOrder.getData()!=null){
                // 下发命令
                PushOrderDto pushOrderDto = new PushOrderDto();
                pushOrderDto.setVin(xpVehicle.getVehicleVin());
                pushOrderDto.setOrderId(newCruiseOrder.getData().getId());
                pushOrderDto.setStartLatitude(newCruiseOrder.getData().getStartLat());
                pushOrderDto.setStartLongitude(newCruiseOrder.getData().getStartLon());
                pushOrderDto.setEndLatitude(newCruiseOrder.getData().getEndLat());
                pushOrderDto.setEndLongitude(newCruiseOrder.getData().getEndLon());
                pushOrderDto.setOrderType(OrderType.CRUISE_ORDER);
                pushOrderDto.setStage(2);
                orderMissionClient.pushOrder(pushOrderDto);
            }


        }else if(dto.getDepartureStatus()==3){
            // 收车 关闭当前巡游单子 生成新的返程单子  找到收车站点
            R<String> r = xpCruiseOrderClient.updateCruiseOrder(xpVehicle.getVehicleVin());

            // redis 获取当前最新经纬度位置
            VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
            XpCruiseOrder xpCruiseOrder = new XpCruiseOrder();
            if(vehiclePosition!=null){
                double x = vehiclePosition.getPoint().getX();
                double y = vehiclePosition.getPoint().getY();
                xpCruiseOrder.setStartLon(x);
                xpCruiseOrder.setStartLat(y);
            }

            xpCruiseOrder.setId("XU"+ IdUtil.fastUUID());
            xpCruiseOrder.setVehicleId(xpVehicle.getId());
            xpCruiseOrder.setFleetId(xpVehicle.getFleetId());
            xpCruiseOrder.setOrderType(2);
            xpCruiseOrder.setStatus(1);
            xpCruiseOrder.setOrderReverse(1);
            xpCruiseOrder.setLastOrderId(r.getData());

            // 找到收车站点
            R<XpCruiseOrder> returnCruisePoint = xpCruisePlanClient.findReturnCruisePoint(xpCruiseOrder);
            if(returnCruisePoint!=null){
                xpCruiseOrderClient.addCruiseOrder(returnCruisePoint.getData());
                // 下发命令
                PushOrderDto pushOrderDto = new PushOrderDto();
                pushOrderDto.setVin(returnCruisePoint.getData().getVehicleVin());
                pushOrderDto.setOrderId(returnCruisePoint.getData().getId());
                pushOrderDto.setStartLatitude(returnCruisePoint.getData().getStartLat());
                pushOrderDto.setStartLongitude(returnCruisePoint.getData().getStartLon());
                pushOrderDto.setEndLatitude(returnCruisePoint.getData().getEndLat());
                pushOrderDto.setEndLongitude(returnCruisePoint.getData().getEndLon());
                pushOrderDto.setOrderType(OrderType.CRUISE_ORDER);
                pushOrderDto.setStage(2);
                orderMissionClient.pushOrder(pushOrderDto);
            }

        }
        return R.ok();
    }

    /**
     * 车辆详情-车辆调度-接单、测试
     */
    @ApiOperation(value = "车辆运维-车辆调度-接单、测试", tags = {"运维APP-车辆管理"})
    @PostMapping("/api/v1/vehicle/vehicleDispatchAccept")
    public R<String> vehicleDispatchAccept(@RequestBody VehicleDispatchAcceptDTO dto) {
        // 查询车辆
        XpVehicle xpVehicle = xpVehicleClient.queryById(dto.getVehicleId()).getData();
        xpVehicle.setAcceptOrTest(dto.getAcceptOrTest());
        xpVehicle.setAcceptOrTestStatus(dto.getAcceptOrTestStatus());
        xpVehicleClient.updateVehicleById(xpVehicle);
        return R.ok();
    }

    /**
     * 车辆详情-车辆调度-运维记录
     */
    @ApiOperation(value = "车辆运维-车辆调度-运维记录", tags = {"运维APP-车辆管理"})
    @PostMapping("/api/v1/vehicle/operationRecords")
    public R<PageInfo<OperationRecordsVO>> operationRecords(@Validated @RequestBody OperationRecordsQuery query) {
        PageInfo<OperationRecordsVO> operationRecords = xpVehiclePreparationService.operationRecords(query);
        return R.ok(operationRecords);
    }

}

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/model/query/XpTaskRecordPageQuery.java

@@ -27,9 +27,11 @@
    @ApiModelProperty(value = "任务类别 1常规 2事故")
    private Integer taskCategory;

    @ApiModelProperty(value = "任务类型id 任务类别为2时对应数据字典,多个逗号分割，为1的时候 枚举1车辆检查 2车辆充电 3车辆整备 4整备洗车 5车辆维修 6车辆保养 7车辆保险到期 8车辆年检到期 9车辆运营资质到期")
    private String taskTypeId;
    @ApiModelProperty(value = "常规任务类型枚举1车辆检查 2车辆充电 3车辆整备 4整备洗车 5车辆维修 6车辆保养 7车辆保险到期 8车辆年检到期 9车辆运营资质到期")
    private Integer taskGeneralId;

    @ApiModelProperty(value = "事故任务类型id")
    private Long taskErrorId;
    @ApiModelProperty(value = "开始时间")
    @JsonFormat(pattern = "yyyy-MM-dd")
    private LocalDate startTime;

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/service/impl/XpTaskServiceImpl.java

@@ -828,10 +828,14 @@
            wrapper.eq(XpTask::getTaskCategory, query.getTaskCategory());
        }

        if (StringUtils.isNotBlank(query.getTaskTypeId())) {
            wrapper.eq(XpTask::getTaskTypeId, query.getTaskTypeId());
        if (Objects.nonNull(query.getTaskGeneralId())) {
            wrapper.eq(XpTask::getTaskTypeId, query.getTaskGeneralId()).eq(XpTask::getTaskCategory, 1);
        }

        if (Objects.nonNull(query.getTaskErrorId())) {
            wrapper.apply("find_in_set({0}, task_type_id)", query.getTaskErrorId())
                    .eq(XpTask::getTaskCategory, 2);
        }
        if (query.getStartTime() != null) {
            wrapper.ge(XpTask::getCreateTime, query.getStartTime().atStartOfDay());
        }

 xp-robotaxi-operation-boot/src/main/java/com/xiaopeng/operation/utils/TaskOrderUtil.java

@@ -2,6 +2,7 @@

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiaopeng.api.dto.automaticPilot.PushOrderDto;
import com.xiaopeng.api.dto.map.RouteAndFareDto;
import com.xiaopeng.api.dto.operation.LonLatDto;
@@ -836,6 +837,17 @@
                        log.info("开始推送 {} 订单到XPU", xpOrder.getOrderNumber());
                        redisService.popFirstElement(quoteName);
                        log.info("将订单 {} 从队列 {} 弹出去", orderNo, quoteName);
                        //判断还原车辆状态
                        Object cacheObject = redisService.getCacheObject("CREATE_ORDER:" + xpOrder.getId().toString());
                        if(null != cacheObject){
                            JSONObject jsonObject = JSON.parseObject(cacheObject.toString());
                            String vin = jsonObject.getString("vin");
                            XpVehicle vehicle = xpVehicleClient.findByVehicleVin(vin).getData();
                            if(null != vehicle && 0 != vehicle.getState()){
                                vehicle.setState(0);
                                xpVehicleClient.updateVehicleById(vehicle);
                            }
                        }
                        return;
                    } catch (Exception e) {
                        e.printStackTrace();

 xp-robotaxi-operation-boot/src/main/resources/mapper/operation/XpElectronicFenceMapper.xml

@@ -20,8 +20,10 @@
        <if test="query.electronicAreaId != null">
            and efa.id = #{query.electronicAreaId}
        </if>
        <if test="query.status != null">
            and ef.status = #{query.status}
        <if test="query.statusList != null">
            <foreach collection="query.statusList" item="status" open="and ef.status in (" separator="," close=")">
                #{status}
            </foreach>
        </if>
        <if test="query.createBy != null and query.createBy!=''">
            and ef.create_by like concat('%',#{query.createBy},'%')

 xp-robotaxi-order-boot/src/main/java/com/xiaopeng/order/controller/XpOrderController.java

@@ -27,6 +27,7 @@
import com.xiaopeng.order.model.dto.XpOrderPayDto;
import com.xiaopeng.order.model.exportExcel.XpOrderExportExcel;
import com.xiaopeng.order.model.query.XpHistoryOrderListQuery;
import com.xiaopeng.order.model.query.XpOrderExcelQuery;
import com.xiaopeng.order.model.query.XpOrderPageQuery;
import com.xiaopeng.order.model.dto.XpOrderOperateDTO;
import com.xiaopeng.order.model.dto.XpOrderReassignDTO;
@@ -229,7 +230,7 @@
     */
    @ApiOperation(value = "订单导出", tags = {"管理后台-交易管理-订单管理"})
    @PostMapping("/api/v1/order/exportExcel")
    public void exportExcel(@RequestBody XpOrderPageQuery query) {
    public void exportExcel(@RequestBody XpOrderExcelQuery query) {
        // 1. 查询导出数据
        List<XpOrderPageVO> dataList = xpOrderService.exportList(query);


 xp-robotaxi-order-boot/src/main/java/com/xiaopeng/order/model/query/XpOrderExcelQuery.java

New file
@@ -0,0 +1,26 @@
package com.xiaopeng.order.model.query;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.xiaopeng.common.core.web.page.BasePage;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import lombok.EqualsAndHashCode;

import java.time.LocalDate;
import java.util.List;

/**
 * 订单分页查询请求
 *
 * @author mitao
 */
@Data
@EqualsAndHashCode(callSuper = true)
@ApiModel(value = "订单分页查询请求", description = "订单分页查询参数")
public class XpOrderExcelQuery extends XpOrderPageQuery {

    @ApiModelProperty(value = "订单ID列表")
    private List<Long> ids;

}

 xp-robotaxi-order-boot/src/main/java/com/xiaopeng/order/service/IXpOrderService.java

@@ -18,6 +18,7 @@
import com.xiaopeng.order.model.dto.XpOrderCommentDto;
import com.xiaopeng.order.model.dto.XpOrderPayDto;
import com.xiaopeng.order.model.query.XpHistoryOrderListQuery;
import com.xiaopeng.order.model.query.XpOrderExcelQuery;
import com.xiaopeng.order.model.query.XpOrderPageQuery;
import com.xiaopeng.order.model.vo.XpOrderDetailVO;
import com.xiaopeng.order.model.vo.XpOrderHistoryVO;
@@ -73,7 +74,7 @@
     * @param query 查询参数
     * @return 订单列表
     */
    List<XpOrderPageVO> exportList(XpOrderPageQuery query);
    List<XpOrderPageVO> exportList(XpOrderExcelQuery query);

    /**
     * 取消或关闭订单

 xp-robotaxi-order-boot/src/main/java/com/xiaopeng/order/service/impl/XpOrderServiceImpl.java

@@ -79,6 +79,7 @@
import com.xiaopeng.order.model.dto.XpOrderCommentDto;
import com.xiaopeng.order.model.dto.XpOrderPayDto;
import com.xiaopeng.order.model.query.XpHistoryOrderListQuery;
import com.xiaopeng.order.model.query.XpOrderExcelQuery;
import com.xiaopeng.order.model.query.XpOrderPageQuery;
import com.xiaopeng.order.model.vo.*;
import com.xiaopeng.order.service.IXpOrderCancelService;
@@ -298,9 +299,9 @@
    }

    @Override
    public List<XpOrderPageVO> exportList(XpOrderPageQuery query) {
    public List<XpOrderPageVO> exportList(XpOrderExcelQuery query) {
        // 1. 构建查询条件
        LambdaQueryWrapper<XpOrder> wrapper = buildQueryWrapper(query);
        LambdaQueryWrapper<XpOrder> wrapper = buildExcelQueryWrapper(query);

        // 2. 查询订单列表
        List<XpOrder> orderList = this.list(wrapper);
@@ -357,7 +358,27 @@
        wrapper.orderByDesc(XpOrder::getCreateTime);
        return wrapper;
    }
    private LambdaQueryWrapper<XpOrder> buildExcelQueryWrapper(XpOrderExcelQuery query) {
        LambdaQueryWrapper<XpOrder> wrapper = new LambdaQueryWrapper<>();

        wrapper.like(StringUtils.isNotBlank(query.getOrderNumber()), XpOrder::getOrderNumber, query.getOrderNumber())
                .eq(Objects.nonNull(query.getFleetId()), XpOrder::getFleetId, query.getFleetId())
                .eq(Objects.nonNull(query.getOrderSource()), XpOrder::getOrderSource, query.getOrderSource())
                .like(StringUtils.isNotBlank(query.getCityCode()), XpOrder::getCityCode, query.getCityCode())
                .eq(Objects.nonNull(query.getStatus()), XpOrder::getStatus, query.getStatus());

        if (Objects.nonNull(query.getStartTime())) {
            wrapper.ge(XpOrder::getCreateTime, LocalDateTime.of(query.getStartTime(), LocalTime.MIN));
        }
        if (Objects.nonNull(query.getEndTime())) {
            wrapper.le(XpOrder::getCreateTime, LocalDateTime.of(query.getEndTime(), LocalTime.MAX));
        }
        if (CollUtil.isNotEmpty(query.getIds())) {
            wrapper.in(XpOrder::getId, query.getIds());
        }
        wrapper.orderByDesc(XpOrder::getCreateTime);
        return wrapper;
    }
    /**
     * 批量获取用户信息
     */
@@ -1016,7 +1037,17 @@
        xpOrderCancel.setCancelSource(1);
        xpOrderCancel.setCancelReason("用户取消订单");
        xpOrderCancelService.save(xpOrderCancel);

        //判断还原车辆状态
        Object cacheObject1 = redisService.getCacheObject("CREATE_ORDER:" + order.getId().toString());
        if(null != cacheObject1){
            com.alibaba.fastjson.JSONObject jsonObject = com.alibaba.fastjson.JSON.parseObject(cacheObject1.toString());
            String vin = jsonObject.getString("vin");
            XpVehicle vehicle = xpVehicleClient.findByVehicleVin(vin).getData();
            if(null != vehicle && 0 != vehicle.getState()){
                vehicle.setState(0);
                xpVehicleClient.updateVehicleById(vehicle);
            }
        }

        // 设置当天的取消次数缓存
        Object cacheObject = redisService.getCacheObject("CANCEL_ORDER_" + uid);

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

@@ -1,27 +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("*");
    }
}
//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/handler/PushMessageWebSocketHandler.java

@@ -1,188 +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());
        }
    }
}
//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,155 +1,175 @@
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 连接管理器
 * 使用 Spring WebSocketSession（共用 Spring Boot HTTP 端口）
 * 支持按用户 ID 管理连接，一个用户可以有多个设备连接
 * 基于AppUser的WebSocket连接管理器
 * 支持按用户ID管理连接，一个用户可以有多个设备连接
 */
@Component
@Slf4j
public class WebSocketUserConnectionManager {

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

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

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

    // ==================== 连接管理 ====================
    private static final ConcurrentHashMap<String, Channel> allConnections = new ConcurrentHashMap<>();

    /**
     * 添加用户连接
     * @param appUserId 用户ID
     * @param channel WebSocket连接通道
     */
    public void addUserConnection(String appUserId, WebSocketSession session) {
        String sessionId = session.getId();
    public void addUserConnection(String appUserId, Channel channel) {
        String channelId = channel.id().asLongText();

        userConnections.computeIfAbsent(appUserId, k -> ConcurrentHashMap.newKeySet()).add(session);
        sessionToUser.put(sessionId, appUserId);
        allConnections.put(sessionId, session);
        // 添加到用户连接映射
        userConnections.computeIfAbsent(appUserId, k -> ConcurrentHashMap.newKeySet())
                .add(channel);

        log.info("用户连接已添加，用户ID: {}, sessionId: {}, 该用户连接数: {}",
                appUserId, sessionId, userConnections.get(appUserId).size());
        // 添加到连接用户映射
        channelToUser.put(channelId, appUserId);

        // 添加到所有连接
        allConnections.put(channelId, channel);

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

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

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

            // 从连接用户映射中移除
            channelToUser.remove(channelId);

            log.info("用户连接已移除，用户ID: {}, ChannelId: {}, 该用户剩余连接数: {}",
                    appUserId, channelId, userChannels != null ? userChannels.size() : 0);
        }

        allConnections.remove(sessionId);
        // 从所有连接中移除
        allConnections.remove(channelId);
    }

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

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

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

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

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

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

    /**
     * 向多个用户发送消息
     * @param appUserIds 用户ID列表
     * @param message 消息内容
     * @return 成功发送的用户数
     */
    public int sendMessageToUsers(List<String> appUserIds, String message) {
        int successUserCount = 0;
@@ -163,57 +183,95 @@
        return successUserCount;
    }

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

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

    /**
     * 获取用户连接数
     * @param appUserId 用户ID
     * @return 连接数
     */
    public int getUserConnectionCount(String appUserId) {
        Set<WebSocketSession> sessions = userConnections.get(appUserId);
        return sessions != null ? sessions.size() : 0;
        Set<Channel> userChannels = userConnections.get(appUserId);
        return userChannels != null ? userChannels.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<WebSocketSession> sessions = userConnections.get(appUserId);
        if (sessions == null || sessions.isEmpty()) {
        Set<Channel> userChannels = userConnections.get(appUserId);
        if (userChannels == null || userChannels.isEmpty()) {
            return false;
        }
        return sessions.stream().anyMatch(WebSocketSession::isOpen);
    }

    public boolean isConnectionExists(String sessionId) {
        WebSocketSession session = allConnections.get(sessionId);
        return session != null && session.isOpen();
        // 检查是否有活跃连接
        return userChannels.stream().anyMatch(Channel::isActive);
    }

    /**
     * 根据 sessionId 获取用户ID
     * 检查连接是否存在
     * @param channelId 连接ID
     * @return 是否存在
     */
    public String getUserIdBySessionId(String sessionId) {
        return sessionToUser.get(sessionId);
    public boolean isConnectionExists(String channelId) {
        Channel channel = allConnections.get(channelId);
        return channel != null && channel.isActive();
    }

    /**
     * 根据连接ID获取用户ID
     * @param channelId 连接ID
     * @return 用户ID
     */
    public String getUserIdByChannelId(String channelId) {
        return channelToUser.get(channelId);
    }

    /**
     * 获取连接统计信息
     * @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<WebSocketSession>> entry : userConnections.entrySet()) {
        for (Map.Entry<String, Set<Channel>> entry : userConnections.entrySet()) {
            userConnectionCounts.put(entry.getKey(), entry.getValue().size());
        }
        stats.put("userConnectionCounts", userConnectionCounts);

 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:
@@ -12,4 +12,150 @@
    allow-circular-references: true
  profiles:
    # 环境配置
    active: local
    active: dev
#    active: prod
---
spring:
  config:
    activate:
      on-profile: dev
  cloud:
    nacos:
      discovery:
        # 服务注册地址
        server-addr: 192.168.110.80:8848
        service: ${spring.application.name}
        group: DEFAULT_GROUP
        namespace: 5185ac4b-6369-48eb-a21b-a7cbc441c5f1
        username: nacos
        password: nacos
      config:
        # 配置中心地址
        server-addr: 192.168.110.80:8848
        namespace: 5185ac4b-6369-48eb-a21b-a7cbc441c5f1
        group: DEFAULT_GROUP
        name: ${spring.application.name}
        username: nacos
        password: nacos
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
---
seata:
  enabled: true
  application-id: ${spring.application.name}
  tx-service-group: seata_tx_group    #此处配置自定义的seata事务分组名称
  enable-auto-data-source-proxy: false    #关闭数据库代理
  service:
    vgroup-mapping:
      seata_tx_group: default
  data-source-proxy-mode: AT
  config:
    type: nacos
    nacos:
      # 开发环境
      server-addr: 192.168.110.80:8848
      namespace: 5185ac4b-6369-48eb-a21b-a7cbc441c5f1
      group: DEFAULT_GROUP
      data-id: seata-server.properties
      username: nacos
      password: nacos
      # 生产环境
  #      server-addr: 192.168.0.137:8848,192.168.0.123:8848
  #      namespace: c6cf40b5-44e8-43f9-be61-8d10fa830e2a
  #      group: DEFAULT_GROUP
  #      data-id: seata-server.properties
  #      username: nacos
  #      password: nacos
  registry:
    type: nacos
    nacos:
      application: seata-server
      # 开发环境
      server-addr: 192.168.110.80:8848
      namespace: 5185ac4b-6369-48eb-a21b-a7cbc441c5f1
      group: DEFAULT_GROUP
      username: nacos
      password: nacos
      # 生产环境
  #      server-addr: 192.168.0.137:8848,192.168.0.123:8848
  #      namespace: c6cf40b5-44e8-43f9-be61-8d10fa830e2a
  #      group: DEFAULT_GROUP
  #      username: nacos
  #      password: nacos
  cloud:
    nacos:
      discovery:
        # 开发环境
        server-addr: 192.168.110.80:8848
        namespace: 5185ac4b-6369-48eb-a21b-a7cbc441c5f1
        group: DEFAULT_GROUP
        application: seata-server #Nacos 中 Seata 名称
        username: nacos
        password: nacos
        # 生产环境
    #        server-addr: 192.168.0.137:8848,192.168.0.123:8848
    #        namespace: c6cf40b5-44e8-43f9-be61-8d10fa830e2a
    #        group: DEFAULT_GROUP
    #        application: seata-server #Nacos 中 Seata 名称
    #        username: nacos
    #        password: nacos
    sentinel:
      transport:
        dashboard: 192.168.0.137:8080  #  Sentinel控制台地址
      eager: true
---
spring:
  config:
    activate:
      on-profile: prod
  cloud:
    nacos:
      discovery:
        # 服务注册地址
        server-addr: 192.168.0.137:8848,192.168.0.123:8848
        service: ${spring.application.name}
        group: DEFAULT_GROUP
        namespace: c6cf40b5-44e8-43f9-be61-8d10fa830e2a
        username: nacos
        password: nacos
      config:
        # 配置中心地址
        server-addr: 192.168.0.137:8848,192.168.0.123:8848
        namespace: c6cf40b5-44e8-43f9-be61-8d10fa830e2a
        username: nacos
        password: nacos
        group: DEFAULT_GROUP
        name: ${spring.application.name}
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
---
spring:
  config:
    activate:
      on-profile: test
  cloud:
    nacos:
      discovery:
        # 服务注册地址
        server-addr: 127.0.0.1:8848
        service: ${spring.application.name}
        group: DEFAULT_GROUP
        namespace: 96712c7a-480b-4f40-b783-39f00f3b33ce
      config:
        # 配置中心地址
        server-addr: 127.0.0.1:8848
        namespace: 96712c7a-480b-4f40-b783-39f00f3b33ce
        group: DEFAULT_GROUP
        name: ${spring.application.name}
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}


 xp-robotaxi-vehicle-convoy-boot/src/main/java/com/xiaopeng/vehicleConvoy/controller/XpFleetController.java

@@ -78,6 +78,21 @@
    }

    /**
     *
     * 根据打卡地址id查询被车辆使用次数
     * @param locationId
     * @return
     */
    @GetMapping("/inner/fleet/getUseCountByLocationId")
    public R<Long> getUseCountByLocationId(@RequestParam Long locationId){

        Long count = xpFleetService.count(Wrappers.lambdaQuery(XpFleet.class)
                .eq(XpFleet::getLocationId, locationId));

        return R.ok(count);
    }

    /**
     * 添加车队
     *
     * @param dto 添加车队DTO

 xp-robotaxi-vehicle-convoy-boot/src/main/java/com/xiaopeng/vehicleConvoy/controller/XpVehicleController.java

@@ -33,6 +33,7 @@
import com.xiaopeng.vehicleConvoy.model.importExcel.XpVehicleImportExcel;
import com.xiaopeng.vehicleConvoy.model.query.*;
import com.xiaopeng.vehicleConvoy.model.vo.XpVehicleVO;
import com.xiaopeng.vehicleConvoy.service.IXpFleetService;
import com.xiaopeng.vehicleConvoy.service.IXpFleetVehicleService;
import com.xiaopeng.vehicleConvoy.service.IXpVehicleAnnualRecordService;
import com.xiaopeng.vehicleConvoy.service.IXpVehicleService;