合并代码

puzhibing

2026-02-13 c3dd800af0f39691a0d13512ce9b2df436d32f41

合并代码

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

@@ -1,155 +1,41 @@
package com.xiaopeng.automaticPilot.controller;

import com.xiaopeng.automaticPilot.service.KafkaProducerService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import com.xiaopeng.automaticPilot.service.KafkaConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.RestController;

/**
 * Kafka 测试控制器
 */

@Slf4j
@RestController
@RequestMapping("/kafka")
@Api(tags = "Kafka 测试接口")
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) {
        try {
            kafkaProducerService.sendMessage(topic, message);
            return "消息发送成功";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-high", groupId = "group-robotaxi-xpu")
    public void topicProdRobotaxiXpuHigh(String message) {
        log.info("收到XPU上报高优先级主题消息: {}", message);
        kafkaConsumerService.topicProdRobotaxiXpuHigh(message);
    }

    /**
     * 发送消息（带 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) {
        try {
            kafkaProducerService.sendMessage(topic, key, message);
            return "消息发送成功";
        } 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();
        }
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-normal", groupId = "group-robotaxi-xpu")
    public void topicProdRobotaxiXpuNormal(String message) {
        log.info("收到XPU上报低优先级主题消息: {}", message);
        kafkaConsumerService.topicProdRobotaxiXpuNormal(message);
    }
}

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

New file
@@ -0,0 +1,31 @@
package com.xiaopeng.automaticPilot.controller;


import com.xiaopeng.automaticPilot.util.XPUUtil;
import com.xiaopeng.common.core.domain.R;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@RestController
@RequestMapping("")
public class XpuVinController {

    @Resource
    private XPUUtil xpuUtil;


    /**
     * 根据车辆VIN查询XPUID
     * @param vin
     * @return
     */
    @GetMapping("/xpu/getXpuIdByVin")
    public R<String> getXpuIdByVin(@RequestParam("vin") String vin){
        String xpuId = xpuUtil.getXpuIdByVin(vin);
        return R.ok(xpuId);
    }
}

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

@@ -89,6 +89,24 @@
     */
    public void topicProdRobotaxiXpuHigh(String message) {
        MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
        handleXPUMessage(msgContent, message);
    }

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


    /**
     * 处理XPU消息
     * @param msgContent
     * @param message
     */
    public void handleXPUMessage(MsgContent msgContent, String message){
        RtMsgHeader header = msgContent.getHeader();
        MessageType type = header.getType();
        //处理请求消息
@@ -173,87 +191,12 @@
                        xpOrderClient.popOrder(dto);
                    }
                }


            }
        }
    }

    /**
     * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
     */
    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);
                    systemStatusXpu(msgContent);
                    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);
                    orderStateUpdateEvent(msgContent);
                    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);
                    navigationDeviationEvent(msgContent);
                    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);
            }


        }

    }


    /**
@@ -273,7 +216,9 @@
        Float alt = naviPose.getAlt();
        //获取车辆静态信息
        VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
        String vin = staticInfo.getVin();
        String xpuId = staticInfo.getXpu_id();
        XpVehicle xpVehicle = xpVehicleClient.findByVehicleXpuId(xpuId).getData();
        String vin = xpVehicle.getVehicleVin();
        //先响应消息
        Long sequence = msgContent.getHeader().getSequence();
        responseMessage(sequence, vin);