KafkaConsumerService的报文传输改成proto生成的文件

zhouhn

2026-02-26 dde1bed7ceaa288aa7d52d468d6f1eafe50db115

 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 com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
import com.xiaopeng.api.dto.operation.LonLatDto;
@@ -18,37 +17,19 @@
import com.xiaopeng.api.model.order.XpOrder;
import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
import com.xiaopeng.api.vo.automaticPilot.VehiclePosition;
import com.xiaopeng.api.vo.automaticPilot.protobuf.ActionType;
import com.xiaopeng.api.vo.automaticPilot.protobuf.GeoPoint;
import com.xiaopeng.api.vo.automaticPilot.protobuf.OrderOperation;
import com.xiaopeng.api.vo.automaticPilot.protobuf.OrderOperationType;
import com.xiaopeng.api.vo.automaticPilot.protobuf.PointType;
import com.xiaopeng.api.vo.automaticPilot.protobuf.ResponseMessage;
import com.xiaopeng.api.vo.automaticPilot.protobuf.RestoreNavigationData;
import com.xiaopeng.api.vo.automaticPilot.protobuf.StatusCode;
import com.xiaopeng.api.vo.automaticPilot.protobuf.Waypoint;
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;
import com.xiaopeng.api.vo.automaticPilot.proto.up.RtMsgHeader;
import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.NavigationDeviationEvent;
import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.NavigationDeviationMessage;
import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.RtEventType;
import com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfoEvent;
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.*;
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest;
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase;
import com.xiaopeng.api.vo.automaticPilot.protobuf.*;
import com.xiaopeng.api.vo.automaticPilot.protobuf.RTCmdType;
import com.google.protobuf.util.JsonFormat;
import com.google.protobuf.InvalidProtocolBufferException;
import com.xiaopeng.api.vo.operation.RouteVo;
import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo;
import com.xiaopeng.automaticPilot.util.XPUUtil;
import com.xiaopeng.common.core.constant.CacheConstants;
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.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.geo.GeoJsonPoint;
import org.springframework.data.mongodb.core.query.Criteria;
@@ -103,16 +84,31 @@
     *
     */
    public void topicProdRobotaxiXpuHigh(String message) {
        MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
        handleXPUMessage(msgContent, message);
        MsgContent msgContent = parseMsgContent(message);
        if (msgContent != null) {
            handleXPUMessage(msgContent, message);
        }
    }

    /**
     * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
     */
    public void topicProdRobotaxiXpuNormal(String message) {
        MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
        handleXPUMessage(msgContent, message);
        MsgContent msgContent = parseMsgContent(message);
        if (msgContent != null) {
            handleXPUMessage(msgContent, message);
        }
    }

    private MsgContent parseMsgContent(String message) {
        try {
            MsgContent.Builder builder = MsgContent.newBuilder();
            JsonFormat.parser().ignoringUnknownFields().merge(message, builder);
            return builder.build();
        } catch (InvalidProtocolBufferException e) {
            log.error("Failed to parse message: {}", message, e);
            return null;
        }
    }


@@ -122,12 +118,17 @@
     * @param message
     */
    public void handleXPUMessage(MsgContent msgContent, String message){
        RtMsgHeader header = msgContent.getHeader();
        MsgHeader header = msgContent.getHeader();
        MessageType type = header.getType();
        //处理请求消息
        if (MessageType.MESSAGE_TYPE_REQUEST == type) {
            //区分业务类型
            RTCmdType cmd = header.getCmd();
            int cmdValue = header.getCmd();
            RTCmdType cmd = RTCmdType.getByValue(cmdValue);
            if (cmd == null) {
                log.warn("Unknown cmd value: {}", cmdValue);
                return;
            }
            switch (cmd) {
                case SYSTEM_STATUS_XPU:
                    log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
@@ -192,10 +193,10 @@
        //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);
            com.xiaopeng.api.vo.automaticPilot.proto.up.StatusCode code = resposne.getCode();
            Long requestSeq = resposne.getRequest_seq();
            if(code == com.xiaopeng.api.vo.automaticPilot.proto.up.StatusCode.OK){
            ResponseMessage response = msgContent.getResponse();
            StatusCode code = response.getCode();
            long requestSeq = response.getRequestSeq();
            if(code == StatusCode.OK){
                Object sequence = redisTemplate.opsForValue().get(requestSeq);
                if(sequence!=null){
                    String[] split = sequence.toString().split("_");
@@ -210,10 +211,6 @@
        }
    }





    /**
     * XPU发布系统运行状态---车辆实时位置
     *
@@ -221,21 +218,21 @@
     */
    public void systemStatusXpu(MsgContent msgContent) {
        //开始处理消息对应的业务逻辑
        SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
        SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus();
        //获取实时位置
        RealtimePose pose = systemRuntimeStatus.getPose();
        Wgs naviPose = pose.getNavi_pose();
        Wgs naviPose = pose.getNaviPose();
        Double latitude = naviPose.getLatitude();
        Double longitude = naviPose.getLongitude();
        Double course = naviPose.getCourse();
        Float alt = naviPose.getAlt();
        //获取车辆静态信息
        VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
        String xpuId = staticInfo.getXpu_id();
        VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicleStatus().getStaticInfo();
        String xpuId = staticInfo.getXpuId();
        XpVehicle xpVehicle = xpVehicleClient.findByVehicleXpuId(xpuId).getData();
        String vin = xpVehicle.getVehicleVin();
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        long sequence = msgContent.getHeader().getSequence();
        responseMessage(sequence, vin);
        //保存车辆实时位置到redis
        redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin);
@@ -249,15 +246,15 @@
        vehiclePosition.setPoint(new GeoJsonPoint(longitude, latitude));
        vehiclePosition.setCourse(course);
        vehiclePosition.setAlt(alt);
        VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicle_status();
        VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicleStatus();
        //车辆动态信息
        VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamic_status();
        dynamicStatus.setSpeed(dynamicStatus.getSpeed());
        vehiclePosition.setCurrentElectricity(dynamicStatus.getBattery_level());
        VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamicStatus();
        vehiclePosition.setSpeed(dynamicStatus.getSpeed());
        vehiclePosition.setCurrentElectricity(dynamicStatus.getDstBatDisp());
        //订单信息
        OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunning_order_status_info();
        OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunningOrderStatusInfo();
        Long orderId = null;
        String order_id = runningOrderStatusInfo.getOrder_id();
        String order_id = runningOrderStatusInfo.getOrderId();
        vehiclePosition.setHasOrder(0);
        if (StringUtils.isNotEmpty(order_id)) {
            if (order_id.contains("XU")) {
@@ -271,14 +268,15 @@
        mongoTemplate.save(vehiclePosition);
        //调用推送服务接口
        if (null != orderId) {
            RtTripInfo rtTripInfo = systemRuntimeStatus.getRt_trip_info();
            RtTripInfo rtTripInfo = systemRuntimeStatus.getRtTripInfo();
            PushOrderRouteDTO dto = new PushOrderRouteDTO();
            dto.setOrderId(orderId);
            dto.setRouteType(order_id.contains("XU") ? 2 : 1);
            dto.setLongitude(longitude);
            dto.setLatitude(latitude);
            dto.setDuration(rtTripInfo.getRemaining_time());
            dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
            //TODO 这两个字段现在拿不到
//            dto.setDuration(rtTripInfo.getRemaining_time());
//            dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
            xpPushMessageClient.pushOrderRoute(dto);
        }
    }
@@ -291,12 +289,13 @@
     */
    public void systemStatusXpuAndReassign(MsgContent msgContent) {
        //开始处理消息对应的业务逻辑
        SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
        AbnormalReason abnormalReason = systemRuntimeStatus.getSystem_status().getAbnormal_reason();
        SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus();
        AbnormalReason abnormalReason = systemRuntimeStatus.getSystemStatus().getAbnormalReason();
        // 存储车辆挡位
        GearLevel gearLevel = systemRuntimeStatus.getVehicle_status().getDynamic_status().getGear_level();
        int value = gearLevel.getValue();
        redisTemplate.opsForValue().set(CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicle_status().getStatic_info().getVin(), value);
        GearLevel gearLevel = systemRuntimeStatus.getVehicleStatus().getDynamicStatus().getGearLevel();
        int value = gearLevel.getNumber();
        //TODO 这里拿不到车架号，只有XPU-ID
//        redisTemplate.opsForValue().set(CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicleStatus().getStaticInfo().getVin(), value);

        // 判断是否为故障状态
        if (abnormalReason == AbnormalReason.FAULT) {
@@ -304,7 +303,7 @@
            // TODO 添加异常任务
            // 解绑用户与车辆关系
            CancelTaxiPassengerVo cancelTaxiPassengerVo = new CancelTaxiPassengerVo();
            String orderId = systemRuntimeStatus.getRunning_order_status_info().getOrder_id();
            String orderId = systemRuntimeStatus.getRunningOrderStatusInfo().getOrderId();
            XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData();
            XpVehicle vehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
            if (vehicle != null) {
@@ -328,21 +327,12 @@
            reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber());
            reassignOrderDTO.setCount(10);
            RealtimePose pose = systemRuntimeStatus.getPose();
            Wgs naviPose = pose.getNavi_pose();
            Wgs naviPose = pose.getNaviPose();
            Double latitude = naviPose.getLatitude();
            Double longitude = naviPose.getLongitude();
            reassignOrderDTO.setLongitude(longitude);
            reassignOrderDTO.setLatitude(latitude);
            reassignOrderClient.reassignOrder(reassignOrderDTO);









//            }
        }
    }
@@ -354,11 +344,11 @@
     */
    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();
        NavigationRequest naviRequest = msgContent.getNavigationRequset();
        String requestId = naviRequest.getRequestId();
        String orderId = naviRequest.getOrderId();
        String taskId = naviRequest.getTaskId();
        OrderPhase orderPhase = naviRequest.getOrderPhase();
        log.info("订单ID：{}", orderId);
        log.info("任务ID：{}", taskId);
        switch (orderPhase) {
@@ -388,17 +378,19 @@
     */
    public void navigationDeviationEvent(MsgContent msgContent) {
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        long sequence = msgContent.getHeader().getSequence();
        //开始处理消息对应的业务逻辑
        NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
        String requestId = deviationEvent.getRequest_id();
        NavigationDeviationEvent deviationEvent = msgContent.getNaviDeviationEvent();

//        String requestId = deviationEvent.getRequestId();
        //偏航事件类型 1：请求导航，2：取消导航
        if (deviationEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
            NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigation_deviation_message();
            String orderId = navigationDeviationMessage.getOrder_id();
            String taskId = navigationDeviationMessage.getTask_id();
            Double longitude = navigationDeviationMessage.getLongitude();
            Double latitude = navigationDeviationMessage.getLatitude();
//        if (deviationEvent.getEventType().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
            NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigationDeviationMessage();
            String orderId = navigationDeviationMessage.getOrderId();
            String taskId = navigationDeviationMessage.getTaskId();
            Wgs wgs = navigationDeviationMessage.getPoint();
            Double longitude = wgs.getLongitude();
            Double latitude = wgs.getLatitude();
            //订单类型  1：乘客订单，2=巡游订单
            LonLatDto destination = new LonLatDto();
            LonLatDto origin = new LonLatDto();
@@ -451,10 +443,11 @@
                    throw new RuntimeException("根据导航ID获取车端需要的导航数据失败");
                }

                //TODO 这里的逻辑不太对，收到导航偏航事件之后
                //构建导航下发请求对象
                cloudPushNavigationInfo(requestId, routeVo.getRouteID(), stringR.getData(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
//                cloudPushNavigationInfo(requestId, routeVo.getRouteID(), stringR.getData(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
            }
        }
//        }
    }


@@ -598,15 +591,15 @@
     */
    public void orderStateUpdateEvent(MsgContent msgContent) {
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        long sequence = msgContent.getHeader().getSequence();
        //开始处理消息对应的业务逻辑
        OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getContent_detail().getObject("orderStatusInfoEvent", OrderStatusInfoEvent.class);
        OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getOrderStatusInfoEvent();
        //只处理请求消息
        if (orderStatusInfoEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
            com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrder_status_info();
            String orderId = orderStatusInfo.getOrder_id();
        if (orderStatusInfoEvent.getEventType().equals(RtEventType.kEventOrderStateUpdate)) {
            OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrderStatusInfo();
            String orderId = orderStatusInfo.getOrderId();
            //普通单
            com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatus status = orderStatusInfo.getStatus();
            OrderStatus status = orderStatusInfo.getStatus();
            switch (status) {
                case ORDER_STATUS_UNSPECIFIED:
                    //订单状态未知