修改分表分库框架和XPU对接逻辑

puzhibing

2026-02-24 907931c9db3d62af425a42decfafc4956a3ddea7

 xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java

@@ -1,7 +1,6 @@
package com.xiaopeng.automaticPilot.service;

import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
@@ -23,6 +22,7 @@
import com.xiaopeng.api.vo.automaticPilot.proto.down.ResponseMessage;
import com.xiaopeng.api.vo.automaticPilot.proto.down.StatusCode;
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.*;
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudPushNavigation.RestoreNavigationData;
import com.xiaopeng.api.vo.automaticPilot.proto.up.MessageType;
import com.xiaopeng.api.vo.automaticPilot.proto.up.MsgContent;
import com.xiaopeng.api.vo.automaticPilot.proto.up.RTCmdType;
@@ -41,10 +41,8 @@
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.enums.OrderStatusEnum;
import com.xiaopeng.common.core.enums.PushMessageTypeEnum;
import com.xiaopeng.common.core.utils.SnowFlake;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
import com.xiaopeng.common.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
@@ -80,15 +78,20 @@
    @Resource
    private XPUUtil xpuUtil;

    @Autowired
    @Resource
    public RedisTemplate redisTemplate;

    @Resource
    private XpPushMessageClient xpPushMessageClient;

    @Resource
    private ReassignOrderClient reassignOrderClient;

    @Resource
    private UserVehicleClient userVehicleClient;

    @Resource
    private Snowflake snowflake;



@@ -182,9 +185,8 @@
                default:
                    log.info("消息不在处理范围内，忽略：{}", message);
            }


        }
        //TODO 这里的逻辑要再确认下
        //消息应答
        if(MessageType.MESSAGE_TYPE_RESPONSE == type){
            com.xiaopeng.api.vo.automaticPilot.proto.up.ResponseMessage resposne = msgContent.getContent_detail().getObject("resposne", com.xiaopeng.api.vo.automaticPilot.proto.up.ResponseMessage.class);
@@ -217,7 +219,7 @@
    public void systemStatusXpu(MsgContent msgContent) {
        //开始处理消息对应的业务逻辑
        SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
        //获取实时位姿
        //获取实时位置
        RealtimePose pose = systemRuntimeStatus.getPose();
        Wgs naviPose = pose.getNavi_pose();
        Double latitude = naviPose.getLatitude();
@@ -333,7 +335,6 @@
        }
    }


    /**
     * XPU向云端请求导航
     *
@@ -342,6 +343,7 @@
    public void xpuNavigationRequest(MsgContent msgContent) {
        //开始处理消息对应的业务逻辑
        NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
        String requestId = naviRequest.getRequest_id();
        String orderId = naviRequest.getOrder_id();
        String taskId = naviRequest.getTask_id();
        OrderPhase orderPhase = naviRequest.getOrder_phase();
@@ -353,15 +355,15 @@
                break;
            case ROUTE_TO_PICKUP:
                log.info("驶往上车点的行程");
                buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
                buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
                break;
            case ROUTE_TO_DESTINATION:
                log.info("驶往下车点的行程");
                buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
                buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
                break;
            case ROUTE_TO_CRUISE:
                log.info("巡游阶段路线");
                buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
                buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
                break;
        }
    }
@@ -377,6 +379,7 @@
        Long sequence = msgContent.getHeader().getSequence();
        //开始处理消息对应的业务逻辑
        NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
        String requestId = deviationEvent.getRequest_id();
        //偏航事件类型 1：请求导航，2：取消导航
        if (deviationEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
            NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigation_deviation_message();
@@ -385,11 +388,9 @@
            Double longitude = navigationDeviationMessage.getLongitude();
            Double latitude = navigationDeviationMessage.getLatitude();
            //订单类型  1：乘客订单，2=巡游订单
            Integer orderType = 1;
            LonLatDto destination = new LonLatDto();
            LonLatDto origin = new LonLatDto();
            if (orderId.contains("XU")) {
                orderType = 2;
                // todo 下发巡游单的导航数据
                //巡游单
                origin.setLongitude(longitude);
@@ -434,7 +435,7 @@
                }
                RouteVo routeVo = routeVoR.getData();
                //构建导航下发请求对象
                cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
                cloudPushNavigationInfo(requestId, routeVo.getRouteID(), routeVo.getRouteDataContent(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
            }
        }
    }
@@ -444,16 +445,13 @@
     * 构建导航请求
     *
     * @param orderId
     * @param taskId
     * @param orderPhase
     */
    public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase, Long sequence) {
    public void buildNaviRequest(String requestId, String orderId, OrderPhase orderPhase, Long sequence) {
        //订单类型  1：乘客订单，2=巡游订单
        Integer orderType = 1;
        LonLatDto destination = new LonLatDto();
        LonLatDto origin = new LonLatDto();
        if (orderId.contains("XU")) {
            orderType = 2;
            // todo 下发巡游单的导航数据
            //巡游单
            if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) {
@@ -507,14 +505,14 @@
            XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
            RouteVo routeVo = routeVoR.getData();
            //构建导航下发请求对象
            cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
            cloudPushNavigationInfo(requestId, routeVo.getRouteID(), routeVo.getRouteDataContent(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
        }
    }


    /**
     * 云端订单操作----下发导航数据
     *
     * 云端订单操作
     * 更新途经点/终点
     * @param routeVo
     */
    public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo, String vin) {
@@ -552,13 +550,11 @@
            waypoints.add(waypoint);
        }
        orderOperation.setWaypoints(waypoints);
        SnowFlake snowFlake = new SnowFlake(0, 0);
        snowFlake.withStartTimestamp(1420041600000L);
        long sequence = snowFlake.nextId();
        long sequence = snowflake.nextId();
        // 处理数据
        redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES);
        //开发发送请求到XPU
        xpuUtil.cloudOrderOperation(orderOperation, vin,sequence);
        xpuUtil.cloudOrderOperation(orderOperation, vin, sequence);
    }


@@ -597,6 +593,7 @@
                case ORDER_RECEIVED:
                    //收到订单
                    buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.READY_DEPART.getCode());
                    break;
                case ORDER_STARTED:
                    //开始执行订单
@@ -605,6 +602,7 @@
                case ORDER_EN_ROUTE_TO_PICKUP:
                    //前往上车点
                    buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP.getCode());
                    break;
                case ORDER_ARRIVED_AT_PICKUP:
                    //到达上车点
@@ -613,14 +611,17 @@
                case ORDER_WAITING_FOR_PICKUP:
                    //等待乘客上车
                    buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode());
                    break;
                case ORDER_PASSENGER_ONBOARD:
                    //乘客已上车
                    buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode());
                    break;
                case ORDER_EN_ROUTE_TO_DESTINATION:
                    //前往目的地
                    buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.IN_SERVICE.getCode());
                    break;
                case ORDER_ARRIVED_AT_DESTINATION:
                    //到达目的地
@@ -629,10 +630,12 @@
                case ORDER_WAITING_FOR_DROPOFF:
                    //等待乘客下车
                    buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
                    break;
                case ORDER_DROPOFF_COMPLETED:
                    //下车完成
                    buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.APPROVAL_REJECTION.getCode());
                    break;
                case ORDER_COMPLETED:
                    //已完成
@@ -641,10 +644,11 @@
                case ORDER_CANCELLED:
                    //已取消
                    buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED, sequence);
                    xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.CANCELED.getCode());
                    break;
                case ORDER_ARRIVED_AT_CRUISE_DESTINATION:
                    //到达巡游终点
                    buildPushOrderStatus(orderId, null, sequence);
                    buildPushOrderStatus(orderId, OrderStatusEnum.COMPLETED, sequence);
                    break;
                case ORDER_PAUSED:
                    //订单暂停
@@ -667,6 +671,15 @@
            //巡游单
            //响应消息
            responseMessage(sequence, "");
            //todo 处理巡游相关的功能
            //前往目的地
            if(OrderStatusEnum.IN_SERVICE.equals(status)){

            }
            //到达目的地
            if(OrderStatusEnum.COMPLETED.equals(status)){

            }
        }else{
            //普通单
            R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
@@ -681,7 +694,7 @@
            content.put("orderId", orderId);
            content.put("status", status.getCode());
            params.put("type", PushMessageTypeEnum.ORDER_STATUS.getDesc());
            params.put("content", com.alibaba.fastjson2.JSONObject.toJSONString(content));
            params.put("content", JSONObject.toJSONString(content));
            params.put("timestamp", System.currentTimeMillis() + "");
            String message = JSONObject.toJSONString(params);
            Long appUserId = xpOrder.getAppUserId();
@@ -695,4 +708,26 @@
        }
    }

    /**
     * 云端下发导航到XPU
     * @param requestId             XPU请求id
     * @param routeId               高德导航id
     * @param routeDataContent      高德导航原数据
     * @param orderId               订单id
     * @param vin                   车辆VIN
     * @param xpuId                 XPU设备id
     */
    public void cloudPushNavigationInfo(String requestId, String routeId, String routeDataContent, String orderId, String vin, String xpuId) {
        //组装报文，将routeDateContent信息推送到XPU
        RestoreNavigationData navigationData = new RestoreNavigationData();
        navigationData.setRoute_id(routeId);
        navigationData.setRequest_id(requestId);
        navigationData.setContent(routeDataContent);
        navigationData.setTimestamp(System.currentTimeMillis());
        long sequence = snowflake.nextId();
        // 处理数据
        redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES);
        xpuUtil.cloudPushNavigation(navigationData, xpuId, sequence);
    }

}