处理启动失败问题

puzhibing

2026-02-12 38858c018592f9788ebc1bd642ceba22bf5cdfae

处理启动失败问题

 xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/KafkaTestController.java

@@ -1,11 +1,12 @@
package com.xiaopeng.automaticPilot.controller;

import com.xiaopeng.automaticPilot.service.KafkaProducerService;
import com.alibaba.fastjson2.JSON;
import com.xiaopeng.api.vo.automaticPilot.proto.up.MsgContent;
import com.xiaopeng.automaticPilot.service.KafkaConsumerService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.*;

/**
@@ -18,138 +19,40 @@
public class KafkaTestController {

    @Autowired
    private KafkaProducerService kafkaProducerService;
    private KafkaConsumerService kafkaConsumerService;




    /**
     * 发送消息（异步）
     * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
     *
     * @param message 消息内容
     */
    @PostMapping("/send")
    @ApiOperation("发送消息（异步）")
    public String sendMessage(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-high", groupId = "group-robotaxi-xpu")
    public void topicProdRobotaxiXpuHigh(String message) {
        try {
            kafkaProducerService.sendMessage(topic, message);
            return "消息发送成功";
            log.info("收到XPU上报高优先级主题消息: {}", message);
            kafkaConsumerService.topicProdRobotaxiXpuHigh(message);
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
            log.error("处理XPU上报高优先级主题消息失败: {}", message, e);
            throw e;
        }
    }

    /**
     * 发送消息（带 Key）
     * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
     *
     * @param message 消息内容
     */
    @PostMapping("/sendWithKey")
    @ApiOperation("发送消息（带 Key）")
    public String sendMessageWithKey(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息 Key", required = true) @RequestParam String key,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-normal", groupId = "group-robotaxi-xpu")
    public void topicProdRobotaxiXpuNormal(String message) {
        try {
            kafkaProducerService.sendMessage(topic, key, message);
            return "消息发送成功";
            log.info("收到XPU上报低优先级主题消息: {}", message);
            kafkaConsumerService.topicProdRobotaxiXpuNormal(message);
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送消息（带回调）
     */
    @PostMapping("/sendWithCallback")
    @ApiOperation("发送消息（带回调）")
    public String sendMessageWithCallback(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessageWithCallback(topic, message);
            return "消息发送成功（带回调）";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送消息（同步）
     */
    @PostMapping("/sendSync")
    @ApiOperation("发送消息（同步）")
    public String sendMessageSync(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessageSync(topic, message);
            return "消息发送成功（同步）";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送自动驾驶测试消息
     */
    @PostMapping("/sendAutoPilot")
    @ApiOperation("发送自动驾驶测试消息")
    public String sendAutoPilotMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("automatic-pilot-topic", message);
            return "自动驾驶消息发送成功";
        } catch (Exception e) {
            log.error("发送自动驾驶消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送车辆状态测试消息
     */
    @PostMapping("/sendVehicleStatus")
    @ApiOperation("发送车辆状态测试消息")
    public String sendVehicleStatusMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("vehicle-status-topic", message);
            return "车辆状态消息发送成功";
        } catch (Exception e) {
            log.error("发送车辆状态消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送导航测试消息
     */
    @PostMapping("/sendNavigation")
    @ApiOperation("发送导航测试消息")
    public String sendNavigationMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("navigation-topic", message);
            return "导航消息发送成功";
        } catch (Exception e) {
            log.error("发送导航消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送订单测试消息
     */
    @PostMapping("/sendOrder")
    @ApiOperation("发送订单测试消息")
    public String sendOrderMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("order-topic", message);
            return "订单消息发送成功";
        } catch (Exception e) {
            log.error("发送订单消息失败", e);
            return "消息发送失败: " + e.getMessage();
            log.error("处理XPU上报低优先级主题消息失败: {}", message, e);
            throw e;
        }
    }
}

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

@@ -1,5 +1,16 @@
package com.xiaopeng.automaticPilot.service;

import com.alibaba.fastjson2.JSON;
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.systemStatusXpu.RealtimePose;
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.SystemRuntimeStatus;
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.VehicleStaticInfo;
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.Wgs;
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest;
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@@ -11,36 +22,213 @@
@Service
public class KafkaConsumerService {




    /**
     * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
     *
     * @param message 消息内容
     */
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-high", groupId = "group-robotaxi-xpu")
    public void consumeAutomaticPilotMessage(String message) {
        try {
            log.info("收到XPU上报高优先级主题消息: {}", message);
        } catch (Exception e) {
            log.error("处理XPU上报高优先级主题消息失败: {}", message, e);
            throw e;
    public void topicProdRobotaxiXpuHigh(String message) {
        MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
        RtMsgHeader header = msgContent.getHeader();
        MessageType type = header.getType();
        //处理请求消息
        if (MessageType.MESSAGE_TYPE_REQUEST == type) {
            //区分业务类型
            RTCmdType cmd = header.getCmd();
            switch (cmd) {
                case SYSTEM_STATUS_XPU:
                    log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
                    break;
                case XPU_NAVIGATION_REQUEST:
                    log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
                    xpuNavigationRequest(msgContent);
                    break;
                case TRIP_STATUS_UPDATE_EVENT:
                    log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PULLOVER_STATUS_EVENT:
                    log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DRIVING_MODE_UPDATE_EVENT:
                    log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case SYSTEM_STATUS_UPDATE_EVENT:
                    log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case TASK_STATE_UPDATE_EVENT:
                    log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_EVENT:
                    log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case POINT_ARRIVAL_STATUS_EVENT:
                    log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PICKUP_CONFIRMATION_EVENT:
                    log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DROPOFF_INFO_EVENT:
                    log.info("用户下车信息(XPU→云端)：{}", message);
                    break;
                case NAVIGATION_DEVIATION_EVENT:
                    log.info("导航偏航(XPU→云端)：{}", message);
                    break;
                case EMERGENCY_STOP_EVENT:
                    log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case NAVIGATION_UPDATE_EVENT:
                    log.info("导航信息变化(XPU→云端)：{}", message);
                    break;
                case TASK_QUEUE_UPDATE_EVENT:
                    log.info("XPU待执行任务队列(XPU→云端)：{}", message);
                    break;
                case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
                    log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_TO_CDU_EVENT:
                    log.info("更新订单状态到CDU (XPU→云端)：{}", message);
                    break;
                default:
                    log.info("消息不在处理范围内，忽略：{}", message);
            }



        }

    }

    /**
     * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
     *
     * @param message 消息内容
     */
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-normal", groupId = "group-robotaxi-xpu")
    public void consumeVehicleStatusMessage(String message) {
        try {
            log.info("收到XPU上报低优先级主题消息: {}", message);
    public void topicProdRobotaxiXpuNormal(String message) {
        MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
        RtMsgHeader header = msgContent.getHeader();
        MessageType type = header.getType();
        //处理请求消息
        if (MessageType.MESSAGE_TYPE_REQUEST == type) {
            //区分业务类型
            RTCmdType cmd = header.getCmd();
            switch (cmd) {
                case SYSTEM_STATUS_XPU:
                    log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
                    break;
                case XPU_NAVIGATION_REQUEST:
                    log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
                    xpuNavigationRequest(msgContent);
                    break;
                case TRIP_STATUS_UPDATE_EVENT:
                    log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PULLOVER_STATUS_EVENT:
                    log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DRIVING_MODE_UPDATE_EVENT:
                    log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case SYSTEM_STATUS_UPDATE_EVENT:
                    log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case TASK_STATE_UPDATE_EVENT:
                    log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_EVENT:
                    log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case POINT_ARRIVAL_STATUS_EVENT:
                    log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PICKUP_CONFIRMATION_EVENT:
                    log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DROPOFF_INFO_EVENT:
                    log.info("用户下车信息(XPU→云端)：{}", message);
                    break;
                case NAVIGATION_DEVIATION_EVENT:
                    log.info("导航偏航(XPU→云端)：{}", message);
                    break;
                case EMERGENCY_STOP_EVENT:
                    log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case NAVIGATION_UPDATE_EVENT:
                    log.info("导航信息变化(XPU→云端)：{}", message);
                    break;
                case TASK_QUEUE_UPDATE_EVENT:
                    log.info("XPU待执行任务队列(XPU→云端)：{}", message);
                    break;
                case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
                    log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_TO_CDU_EVENT:
                    log.info("更新订单状态到CDU (XPU→云端)：{}", message);
                    break;
                default:
                    log.info("消息不在处理范围内，忽略：{}", message);
            }

        } catch (Exception e) {
            log.error("处理XPU上报低优先级主题消息失败: {}", message, e);
            throw e;

        }

    }


    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();
        Double longitude = naviPose.getLongitude();
        Double course = naviPose.getCourse();
        Float alt = naviPose.getAlt();
        //获取车辆静态信息
        VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
        String vin = staticInfo.getVin();



    }





    /**
     * XPU向云端请求导航
     * @param msgContent
     */
    public void xpuNavigationRequest(MsgContent msgContent){
        NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
        String orderId = naviRequest.getOrder_id();
        String taskId = naviRequest.getTask_id();
        OrderPhase orderPhase = naviRequest.getOrder_phase();
        log.info("订单ID：{}", orderId);
        log.info("任务ID：{}", taskId);
        switch (orderPhase){
            case IDLE:
                log.info("订单的任务未开始执行");
                break;
            case ROUTE_TO_PICKUP:
                log.info("驶往上车点的行程");

                break;
            case ROUTE_TO_DESTINATION:
                log.info("驶往下车点的行程");
                break;
            case ROUTE_TO_CRUISE:
                log.info("巡游阶段路线");
                break;
        }
    }



    public void c(MsgContent msgContent){



    }

}

 xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaProducerService.java

File was deleted