From 38858c018592f9788ebc1bd642ceba22bf5cdfae Mon Sep 17 00:00:00 2001
From: puzhibing <393733352@qq.com>
Date: 星期四, 12 二月 2026 18:50:14 +0800
Subject: [PATCH] 处理启动失败问题
---
xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/KafkaTestController.java | 149 ++++++++-----------------------------------------
1 files changed, 26 insertions(+), 123 deletions(-)
diff --git a/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/KafkaTestController.java b/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/KafkaTestController.java
index f8138e5..5cd28b5 100644
--- a/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/KafkaTestController.java
+++ b/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;
}
}
}
--
Gitblit v1.7.1