From 907931c9db3d62af425a42decfafc4956a3ddea7 Mon Sep 17 00:00:00 2001
From: puzhibing <393733352@qq.com>
Date: 星期二, 24 二月 2026 18:07:48 +0800
Subject: [PATCH] 修改分表分库框架和XPU对接逻辑
---
xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 89 +++++++++++++++++++++++++++++++-------------
1 files changed, 62 insertions(+), 27 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 72cd1e9..0efb36b 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 cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
@@ -23,6 +22,7 @@
import com.xiaopeng.api.vo.automaticPilot.proto.down.ResponseMessage;
import com.xiaopeng.api.vo.automaticPilot.proto.down.StatusCode;
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.*;
+import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudPushNavigation.RestoreNavigationData;
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;
@@ -41,10 +41,8 @@
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.SnowFlake;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
-import com.xiaopeng.common.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
@@ -80,15 +78,20 @@
@Resource
private XPUUtil xpuUtil;
- @Autowired
+ @Resource
public RedisTemplate redisTemplate;
@Resource
private XpPushMessageClient xpPushMessageClient;
+
@Resource
private ReassignOrderClient reassignOrderClient;
+
@Resource
private UserVehicleClient userVehicleClient;
+
+ @Resource
+ private Snowflake snowflake;
@@ -182,9 +185,8 @@
default:
log.info("消息不在处理范围内，忽略：{}", message);
}
-
-
}
+ //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);
@@ -217,7 +219,7 @@
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();
@@ -333,7 +335,6 @@
}
}
-
/**
* XPU向云端请求导航
*
@@ -342,6 +343,7 @@
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();
@@ -353,15 +355,15 @@
break;
case ROUTE_TO_PICKUP:
log.info("驶往上车点的行程");
- buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
+ buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
break;
case ROUTE_TO_DESTINATION:
log.info("驶往下车点的行程");
- buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
+ buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
break;
case ROUTE_TO_CRUISE:
log.info("巡游阶段路线");
- buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
+ buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence());
break;
}
}
@@ -377,6 +379,7 @@
Long sequence = msgContent.getHeader().getSequence();
//开始处理消息对应的业务逻辑
NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
+ String requestId = deviationEvent.getRequest_id();
//偏航事件类型 1：请求导航，2：取消导航
if (deviationEvent.getEvent_type().equals(RtEventType.RT_EVENT_TYPE_REQUEST)) {
NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigation_deviation_message();
@@ -385,11 +388,9 @@
Double longitude = navigationDeviationMessage.getLongitude();
Double latitude = navigationDeviationMessage.getLatitude();
//订单类型 1：乘客订单，2=巡游订单
- Integer orderType = 1;
LonLatDto destination = new LonLatDto();
LonLatDto origin = new LonLatDto();
if (orderId.contains("XU")) {
- orderType = 2;
// todo 下发巡游单的导航数据
//巡游单
origin.setLongitude(longitude);
@@ -434,7 +435,7 @@
}
RouteVo routeVo = routeVoR.getData();
//构建导航下发请求对象
- cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
+ cloudPushNavigationInfo(requestId, routeVo.getRouteID(), routeVo.getRouteDataContent(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
}
}
}
@@ -444,16 +445,13 @@
* 构建导航请求
*
* @param orderId
- * @param taskId
* @param orderPhase
*/
- public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase, Long sequence) {
+ public void buildNaviRequest(String requestId, String orderId, OrderPhase orderPhase, Long sequence) {
//订单类型 1：乘客订单，2=巡游订单
- Integer orderType = 1;
LonLatDto destination = new LonLatDto();
LonLatDto origin = new LonLatDto();
if (orderId.contains("XU")) {
- orderType = 2;
// todo 下发巡游单的导航数据
//巡游单
if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) {
@@ -507,14 +505,14 @@
XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
RouteVo routeVo = routeVoR.getData();
//构建导航下发请求对象
- cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
+ cloudPushNavigationInfo(requestId, routeVo.getRouteID(), routeVo.getRouteDataContent(), orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId());
}
}
/**
- * 云端订单操作----下发导航数据
- *
+ * 云端订单操作
+ * 更新途经点/终点
* @param routeVo
*/
public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo, String vin) {
@@ -552,13 +550,11 @@
waypoints.add(waypoint);
}
orderOperation.setWaypoints(waypoints);
- SnowFlake snowFlake = new SnowFlake(0, 0);
- snowFlake.withStartTimestamp(1420041600000L);
- long sequence = snowFlake.nextId();
+ long sequence = snowflake.nextId();
// 处理数据
redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES);
//开发发送请求到XPU
- xpuUtil.cloudOrderOperation(orderOperation, vin,sequence);
+ xpuUtil.cloudOrderOperation(orderOperation, vin, sequence);
}
@@ -597,6 +593,7 @@
case ORDER_RECEIVED:
//收到订单
buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.READY_DEPART.getCode());
break;
case ORDER_STARTED:
//开始执行订单
@@ -605,6 +602,7 @@
case ORDER_EN_ROUTE_TO_PICKUP:
//前往上车点
buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP.getCode());
break;
case ORDER_ARRIVED_AT_PICKUP:
//到达上车点
@@ -613,14 +611,17 @@
case ORDER_WAITING_FOR_PICKUP:
//等待乘客上车
buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode());
break;
case ORDER_PASSENGER_ONBOARD:
//乘客已上车
buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode());
break;
case ORDER_EN_ROUTE_TO_DESTINATION:
//前往目的地
buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.IN_SERVICE.getCode());
break;
case ORDER_ARRIVED_AT_DESTINATION:
//到达目的地
@@ -629,10 +630,12 @@
case ORDER_WAITING_FOR_DROPOFF:
//等待乘客下车
buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
break;
case ORDER_DROPOFF_COMPLETED:
//下车完成
buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.APPROVAL_REJECTION.getCode());
break;
case ORDER_COMPLETED:
//已完成
@@ -641,10 +644,11 @@
case ORDER_CANCELLED:
//已取消
buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.CANCELED.getCode());
break;
case ORDER_ARRIVED_AT_CRUISE_DESTINATION:
//到达巡游终点
- buildPushOrderStatus(orderId, null, sequence);
+ buildPushOrderStatus(orderId, OrderStatusEnum.COMPLETED, sequence);
break;
case ORDER_PAUSED:
//订单暂停
@@ -667,6 +671,15 @@
//巡游单
//响应消息
responseMessage(sequence, "");
+ //todo 处理巡游相关的功能
+ //前往目的地
+ if(OrderStatusEnum.IN_SERVICE.equals(status)){
+
+ }
+ //到达目的地
+ if(OrderStatusEnum.COMPLETED.equals(status)){
+
+ }
}else{
//普通单
R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
@@ -681,7 +694,7 @@
content.put("orderId", orderId);
content.put("status", status.getCode());
params.put("type", PushMessageTypeEnum.ORDER_STATUS.getDesc());
- params.put("content", com.alibaba.fastjson2.JSONObject.toJSONString(content));
+ params.put("content", JSONObject.toJSONString(content));
params.put("timestamp", System.currentTimeMillis() + "");
String message = JSONObject.toJSONString(params);
Long appUserId = xpOrder.getAppUserId();
@@ -695,4 +708,26 @@
}
}
+ /**
+ * 云端下发导航到XPU
+ * @param requestId XPU请求id
+ * @param routeId 高德导航id
+ * @param routeDataContent 高德导航原数据
+ * @param orderId 订单id
+ * @param vin 车辆VIN
+ * @param xpuId XPU设备id
+ */
+ public void cloudPushNavigationInfo(String requestId, String routeId, String routeDataContent, String orderId, String vin, String xpuId) {
+ //组装报文，将routeDateContent信息推送到XPU
+ RestoreNavigationData navigationData = new RestoreNavigationData();
+ navigationData.setRoute_id(routeId);
+ navigationData.setRequest_id(requestId);
+ navigationData.setContent(routeDataContent);
+ navigationData.setTimestamp(System.currentTimeMillis());
+ long sequence = snowflake.nextId();
+ // 处理数据
+ redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES);
+ xpuUtil.cloudPushNavigation(navigationData, xpuId, sequence);
+ }
+
}
--
Gitblit v1.7.1