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

liujie

2026-02-12 2064bc46d0d716a3a6eb1ab9c113ee8b7748a613

 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;
        }
    }
}