From dde1bed7ceaa288aa7d52d468d6f1eafe50db115 Mon Sep 17 00:00:00 2001
From: zhouhn <zhouhn@xiaopeng.com>
Date: 星期六, 07 三月 2026 20:59:19 +0800
Subject: [PATCH] KafkaConsumerService的报文传输改成proto生成的文件
---
xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 171 +++++++++++++++++++++++++++-----------------------------
1 files changed, 82 insertions(+), 89 deletions(-)
diff --git a/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java b/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
index a0fec39..85d2979 100644
--- a/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
+++ b/xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
@@ -1,7 +1,6 @@
package com.xiaopeng.automaticPilot.service;
import cn.hutool.core.lang.Snowflake;
-import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
import com.xiaopeng.api.dto.operation.LonLatDto;
@@ -18,37 +17,19 @@
import com.xiaopeng.api.model.order.XpOrder;
import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
import com.xiaopeng.api.vo.automaticPilot.VehiclePosition;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.ActionType;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.GeoPoint;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.OrderOperation;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.OrderOperationType;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.PointType;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.ResponseMessage;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.RestoreNavigationData;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.StatusCode;
-import com.xiaopeng.api.vo.automaticPilot.protobuf.Waypoint;
-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.navigationDeviationEvent.NavigationDeviationEvent;
-import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.NavigationDeviationMessage;
-import com.xiaopeng.api.vo.automaticPilot.proto.up.navigationDeviationEvent.RtEventType;
-import com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfoEvent;
-import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.*;
-import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest;
-import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase;
+import com.xiaopeng.api.vo.automaticPilot.protobuf.*;
+import com.xiaopeng.api.vo.automaticPilot.protobuf.RTCmdType;
+import com.google.protobuf.util.JsonFormat;
+import com.google.protobuf.InvalidProtocolBufferException;
import com.xiaopeng.api.vo.operation.RouteVo;
import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo;
import com.xiaopeng.automaticPilot.util.XPUUtil;
-import com.xiaopeng.common.core.constant.CacheConstants;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.enums.OrderStatusEnum;
import com.xiaopeng.common.core.enums.PushMessageTypeEnum;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.geo.GeoJsonPoint;
import org.springframework.data.mongodb.core.query.Criteria;
@@ -103,16 +84,31 @@
*
*/
public void topicProdRobotaxiXpuHigh(String message) {
- MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
- handleXPUMessage(msgContent, message);
+ MsgContent msgContent = parseMsgContent(message);
+ if (msgContent != null) {
+ handleXPUMessage(msgContent, message);
+ }
}
/**
* 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
*/
public void topicProdRobotaxiXpuNormal(String message) {
- MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
- handleXPUMessage(msgContent, message);
+ MsgContent msgContent = parseMsgContent(message);
+ if (msgContent != null) {
+ handleXPUMessage(msgContent, message);
+ }
+ }
+
+ private MsgContent parseMsgContent(String message) {
+ try {
+ MsgContent.Builder builder = MsgContent.newBuilder();
+ JsonFormat.parser().ignoringUnknownFields().merge(message, builder);
+ return builder.build();
+ } catch (InvalidProtocolBufferException e) {
+ log.error("Failed to parse message: {}", message, e);
+ return null;
+ }
}
@@ -122,12 +118,17 @@
* @param message
*/
public void handleXPUMessage(MsgContent msgContent, String message){
- RtMsgHeader header = msgContent.getHeader();
+ MsgHeader header = msgContent.getHeader();
MessageType type = header.getType();
//处理请求消息
if (MessageType.MESSAGE_TYPE_REQUEST == type) {
//区分业务类型
- RTCmdType cmd = header.getCmd();
+ int cmdValue = header.getCmd();
+ RTCmdType cmd = RTCmdType.getByValue(cmdValue);
+ if (cmd == null) {
+ log.warn("Unknown cmd value: {}", cmdValue);
+ return;
+ }
switch (cmd) {
case SYSTEM_STATUS_XPU:
log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
@@ -192,10 +193,10 @@
//TODO 这里的逻辑要再确认下
//消息应答
if(MessageType.MESSAGE_TYPE_RESPONSE == type){
- com.xiaopeng.api.vo.automaticPilot.proto.up.ResponseMessage resposne = msgContent.getContent_detail().getObject("resposne", com.xiaopeng.api.vo.automaticPilot.proto.up.ResponseMessage.class);
- com.xiaopeng.api.vo.automaticPilot.proto.up.StatusCode code = resposne.getCode();
- Long requestSeq = resposne.getRequest_seq();
- if(code == com.xiaopeng.api.vo.automaticPilot.proto.up.StatusCode.OK){
+ ResponseMessage response = msgContent.getResponse();
+ StatusCode code = response.getCode();
+ long requestSeq = response.getRequestSeq();
+ if(code == StatusCode.OK){
Object sequence = redisTemplate.opsForValue().get(requestSeq);
if(sequence!=null){
String[] split = sequence.toString().split("_");
@@ -210,10 +211,6 @@
}
}
-
-
-
-
/**
* XPU发布系统运行状态---车辆实时位置
*
@@ -221,21 +218,21 @@
*/
public void systemStatusXpu(MsgContent msgContent) {
//开始处理消息对应的业务逻辑
- SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
+ SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus();
//获取实时位置
RealtimePose pose = systemRuntimeStatus.getPose();
- Wgs naviPose = pose.getNavi_pose();
+ Wgs naviPose = pose.getNaviPose();
Double latitude = naviPose.getLatitude();
Double longitude = naviPose.getLongitude();
Double course = naviPose.getCourse();
Float alt = naviPose.getAlt();
//获取车辆静态信息
- VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
- String xpuId = staticInfo.getXpu_id();
+ VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicleStatus().getStaticInfo();
+ String xpuId = staticInfo.getXpuId();
XpVehicle xpVehicle = xpVehicleClient.findByVehicleXpuId(xpuId).getData();
String vin = xpVehicle.getVehicleVin();
//先响应消息
- Long sequence = msgContent.getHeader().getSequence();
+ long sequence = msgContent.getHeader().getSequence();
responseMessage(sequence, vin);
//保存车辆实时位置到redis
redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin);
@@ -249,15 +246,15 @@
vehiclePosition.setPoint(new GeoJsonPoint(longitude, latitude));
vehiclePosition.setCourse(course);
vehiclePosition.setAlt(alt);
- VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicle_status();
+ VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicleStatus();
//车辆动态信息
- VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamic_status();
- dynamicStatus.setSpeed(dynamicStatus.getSpeed());
- vehiclePosition.setCurrentElectricity(dynamicStatus.getBattery_level());
+ VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamicStatus();
+ vehiclePosition.setSpeed(dynamicStatus.getSpeed());
+ vehiclePosition.setCurrentElectricity(dynamicStatus.getDstBatDisp());
//订单信息
- OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunning_order_status_info();
+ OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunningOrderStatusInfo();
Long orderId = null;
- String order_id = runningOrderStatusInfo.getOrder_id();
+ String order_id = runningOrderStatusInfo.getOrderId();
vehiclePosition.setHasOrder(0);
if (StringUtils.isNotEmpty(order_id)) {
if (order_id.contains("XU")) {
@@ -271,14 +268,15 @@
mongoTemplate.save(vehiclePosition);
//调用推送服务接口
if (null != orderId) {
- RtTripInfo rtTripInfo = systemRuntimeStatus.getRt_trip_info();
+ RtTripInfo rtTripInfo = systemRuntimeStatus.getRtTripInfo();
PushOrderRouteDTO dto = new PushOrderRouteDTO();
dto.setOrderId(orderId);
dto.setRouteType(order_id.contains("XU") ? 2 : 1);
dto.setLongitude(longitude);
dto.setLatitude(latitude);
- dto.setDuration(rtTripInfo.getRemaining_time());
- dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
+ //TODO 这两个字段现在拿不到
+// dto.setDuration(rtTripInfo.getRemaining_time());
+// dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
xpPushMessageClient.pushOrderRoute(dto);
}
}
@@ -291,12 +289,13 @@
*/
public void systemStatusXpuAndReassign(MsgContent msgContent) {
//开始处理消息对应的业务逻辑
- SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
- AbnormalReason abnormalReason = systemRuntimeStatus.getSystem_status().getAbnormal_reason();
+ SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus();
+ AbnormalReason abnormalReason = systemRuntimeStatus.getSystemStatus().getAbnormalReason();
// 存储车辆挡位
- GearLevel gearLevel = systemRuntimeStatus.getVehicle_status().getDynamic_status().getGear_level();
- int value = gearLevel.getValue();
- redisTemplate.opsForValue().set(CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicle_status().getStatic_info().getVin(), value);
+ GearLevel gearLevel = systemRuntimeStatus.getVehicleStatus().getDynamicStatus().getGearLevel();
+ int value = gearLevel.getNumber();
+ //TODO 这里拿不到车架号，只有XPU-ID
+// redisTemplate.opsForValue().set(CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicleStatus().getStaticInfo().getVin(), value);
// 判断是否为故障状态
if (abnormalReason == AbnormalReason.FAULT) {
@@ -304,7 +303,7 @@
// TODO 添加异常任务
// 解绑用户与车辆关系
CancelTaxiPassengerVo cancelTaxiPassengerVo = new CancelTaxiPassengerVo();
- String orderId = systemRuntimeStatus.getRunning_order_status_info().getOrder_id();
+ String orderId = systemRuntimeStatus.getRunningOrderStatusInfo().getOrderId();
XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData();
XpVehicle vehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
if (vehicle != null) {
@@ -328,21 +327,12 @@
reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber());
reassignOrderDTO.setCount(10);
RealtimePose pose = systemRuntimeStatus.getPose();
- Wgs naviPose = pose.getNavi_pose();
+ Wgs naviPose = pose.getNaviPose();
Double latitude = naviPose.getLatitude();
Double longitude = naviPose.getLongitude();
reassignOrderDTO.setLongitude(longitude);
reassignOrderDTO.setLatitude(latitude);
reassignOrderClient.reassignOrder(reassignOrderDTO);
-
-
-
-
-
-
-
-
-
// }
}
}
@@ -354,11 +344,11 @@
*/
public void xpuNavigationRequest(MsgContent msgContent) {
//开始处理消息对应的业务逻辑
- NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
- String requestId = naviRequest.getRequest_id();
- String orderId = naviRequest.getOrder_id();
- String taskId = naviRequest.getTask_id();
- OrderPhase orderPhase = naviRequest.getOrder_phase();
+ NavigationRequest naviRequest = msgContent.getNavigationRequset();
+ String requestId = naviRequest.getRequestId();
+ String orderId = naviRequest.getOrderId();
+ String taskId = naviRequest.getTaskId();
+ OrderPhase orderPhase = naviRequest.getOrderPhase();
log.info("订单ID：{}", orderId);
log.info("任务ID：{}", taskId);
switch (orderPhase) {
@@ -388,17 +378,19 @@
*/
public void navigationDeviationEvent(MsgContent msgContent) {
//先响应消息
- Long sequence = msgContent.getHeader().getSequence();
+ long sequence = msgContent.getHeader().getSequence();
//开始处理消息对应的业务逻辑
- NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
- String requestId = deviationEvent.getRequest_id();
+ NavigationDeviationEvent deviationEvent = msgContent.getNaviDeviationEvent();
+
+// String requestId = deviationEvent.getRequestId();
//偏航事件类型 1：请求导航，2：取消导航
- if (deviationEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
- NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigation_deviation_message();
- String orderId = navigationDeviationMessage.getOrder_id();
- String taskId = navigationDeviationMessage.getTask_id();
- Double longitude = navigationDeviationMessage.getLongitude();
- Double latitude = navigationDeviationMessage.getLatitude();
+// if (deviationEvent.getEventType().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
+ NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigationDeviationMessage();
+ String orderId = navigationDeviationMessage.getOrderId();
+ String taskId = navigationDeviationMessage.getTaskId();
+ Wgs wgs = navigationDeviationMessage.getPoint();
+ Double longitude = wgs.getLongitude();
+ Double latitude = wgs.getLatitude();
//订单类型 1：乘客订单，2=巡游订单
LonLatDto destination = new LonLatDto();
LonLatDto origin = new LonLatDto();
@@ -451,10 +443,11 @@
throw new RuntimeException("根据导航ID获取车端需要的导航数据失败");
}
+ //TODO 这里的逻辑不太对，收到导航偏航事件之后
//构建导航下发请求对象
- cloudPushNavigationInfo(requestId, routeVo.getRouteID(), stringR.getData(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
+// cloudPushNavigationInfo(requestId, routeVo.getRouteID(), stringR.getData(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
}
- }
+// }
}
@@ -598,15 +591,15 @@
*/
public void orderStateUpdateEvent(MsgContent msgContent) {
//先响应消息
- Long sequence = msgContent.getHeader().getSequence();
+ long sequence = msgContent.getHeader().getSequence();
//开始处理消息对应的业务逻辑
- OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getContent_detail().getObject("orderStatusInfoEvent", OrderStatusInfoEvent.class);
+ OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getOrderStatusInfoEvent();
//只处理请求消息
- if (orderStatusInfoEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
- com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrder_status_info();
- String orderId = orderStatusInfo.getOrder_id();
+ if (orderStatusInfoEvent.getEventType().equals(RtEventType.kEventOrderStateUpdate)) {
+ OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrderStatusInfo();
+ String orderId = orderStatusInfo.getOrderId();
//普通单
- com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatus status = orderStatusInfo.getStatus();
+ OrderStatus status = orderStatusInfo.getStatus();
switch (status) {
case ORDER_STATUS_UNSPECIFIED:
//订单状态未知
--
Gitblit v1.7.1