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/service/KafkaConsumerService.java | 222 +++++++++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 205 insertions(+), 17 deletions(-)
diff --git a/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java b/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
index a1593f8..d598a8b 100644
--- a/xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
+++ b/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){
+
+
+
+ }
+
}
--
Gitblit v1.7.1