From 9adef3dd7723930d44f72713c3cf28e50b002055 Mon Sep 17 00:00:00 2001
From: liujie <1793218484@qq.com>
Date: 星期三, 01 四月 2026 15:45:19 +0800
Subject: [PATCH] Merge branch 'master' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi
---
xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 303 ++++++++++++++++++++++++++++++++++++++-----------
1 files changed, 233 insertions(+), 70 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 73ad12f..b79d9e2 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
@@ -42,6 +42,7 @@
import com.xiaopeng.common.core.constant.CacheConstants;
import com.xiaopeng.common.core.constant.Constants;
import com.xiaopeng.common.core.domain.R;
+import com.xiaopeng.common.core.enums.CancelOrderSceneEnum;
import com.xiaopeng.common.core.enums.OrderStatusEnum;
import com.xiaopeng.common.core.enums.PushMessageTypeEnum;
import com.xiaopeng.common.core.utils.StringUtils;
@@ -71,6 +72,8 @@
@Slf4j
@Service
public class KafkaConsumerService {
+ private static final String CANCEL_ORDER_CONTEXT_PREFIX = "CANCEL_ORDER_CONTEXT:";
+
@Resource
private XpMapClient xpMapClient;
@@ -441,6 +444,7 @@
}
}else if(orderId.contains("TEST")){
//测试订单故障则取消订单
+ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), 0);
XpTestOrder testOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4))).getData();
vehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData();
if (vehicle != null) {
@@ -591,6 +595,79 @@
vehicleId = xpCruiseOrder.getVehicleId();
+ } else if(orderId.contains("TEST")){
+ R<XpTestOrder> cruiseOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4)));
+ XpTestOrder testOrder = cruiseOrder.getData();
+ //区分测试单和测试巡游单
+ if(1 == testOrder.getOrderType()){
+ orderType = OrderType.TEST_ORDER;
+ //驶往上车点的行程
+ if (3 == testOrder.getStatus()) {
+ //查询司机当前位置到起点的线路
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(testOrder.getStartLon());
+ destination.setLatitude(testOrder.getStartLat());
+
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId.substring(4)));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ dto.setRouteType(1);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ routeID = routeVo.getRouteID();
+ //缓存数据，用于后续推给乘客导航线路
+ redisService.setCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + testOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
+ }
+ //驶往下车点的行程
+ if (6 == testOrder.getStatus()) {
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(testOrder.getEndLon());
+ destination.setLatitude(testOrder.getEndLat());
+
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId.substring(4)));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ dto.setRouteType(2);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ routeID = routeVo.getRouteID();
+ //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据
+ redisService.setCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + testOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
+ }
+
+ }else{
+ orderType = OrderType.TEST_CRUISE_ORDER;
+
+ //构建导航数据给到车端
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(testOrder.getEndLon());
+ destination.setLatitude(testOrder.getEndLat());
+
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId.substring(4)));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ dto.setRouteType(3);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ routeID = routeVo.getRouteID();
+
+ vehicleId = testOrder.getVehicleId();
+ }
} else {
orderType = OrderType.USER_ORDER;
//正常单
@@ -701,77 +778,100 @@
vehicleId = xpCruiseOrder.getVehicleId();
}
} else if(orderId.contains("TEST")){
- orderType = OrderType.TEST_ORDER;
- //正常单
- R<XpTestOrder> xpOrderR = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4)));
- if (200 != xpOrderR.getCode()) {
- throw new RuntimeException("订单不存在 " + orderId);
- }
- XpTestOrder xpOrder = xpOrderR.getData();
- XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
- //驶往上车点的行程
- if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) {
- //直接取缓存中的导航数据推送给车端和乘客
- Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + xpOrder.getVehicleId());
- if(null == cacheObject){
- //没有缓存数据，重新规划线路
- //查询车辆位置
- VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
- if (Objects.isNull(vehiclePosition)) {
- throw new RuntimeException("车辆位置信息不存在");
+ R<XpTestOrder> cruiseOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4)));
+ XpTestOrder testOrder = cruiseOrder.getData();
+ XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData();
+ //区分测试单和测试巡游单
+ if(1 == testOrder.getOrderType()){
+ orderType = OrderType.TEST_ORDER;
+ //驶往上车点的行程
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) {
+ //直接取缓存中的导航数据推送给车端和乘客
+ Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + testOrder.getVehicleId());
+ if(null == cacheObject){
+ //没有缓存数据，重新规划线路
+ //查询车辆位置
+ VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
+ if (Objects.isNull(vehiclePosition)) {
+ throw new RuntimeException("车辆位置信息不存在");
+ }
+ origin.setLongitude(vehiclePosition.getPoint().getX());
+ origin.setLatitude(vehiclePosition.getPoint().getY());
+ destination.setLongitude(testOrder.getStartLon());
+ destination.setLatitude(testOrder.getStartLat());
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId.substring(4)));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ dto.setRouteType(2);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ routeID = routeVo.getRouteID();
+ //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据
+ redisService.setCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + testOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
+ }else{
+ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class);
+ routeID = routeVo.getRouteID();
}
- origin.setLongitude(vehiclePosition.getPoint().getX());
- origin.setLatitude(vehiclePosition.getPoint().getY());
- destination.setLongitude(xpOrder.getStartLon());
- destination.setLatitude(xpOrder.getStartLat());
+ }
+ //驶往下车点的行程
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) {
+ //直接取缓存中的导航数据推送给车端和乘客
+ Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + testOrder.getVehicleId());
+ if(null == cacheObject){
+ //没有缓存数据，重新规划线路
+ //查询起点到终点的线路
+ origin.setLongitude(testOrder.getStartLon());
+ origin.setLatitude(testOrder.getStartLat());
+ destination.setLongitude(testOrder.getEndLon());
+ destination.setLatitude(testOrder.getEndLat());
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId.substring(4)));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ dto.setRouteType(2);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ routeID = routeVo.getRouteID();
+ //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据
+ redisService.setCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + testOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
+ }else{
+ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class);
+ routeID = routeVo.getRouteID();
+ }
+ }
+
+ }else{
+ orderType = OrderType.TEST_CRUISE_ORDER;
+
+ //巡游单
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) {
+ /* 构建导航数据给到车端 */
+ origin.setLongitude(testOrder.getStartLon());
+ origin.setLatitude(testOrder.getStartLat());
+ destination.setLongitude(testOrder.getEndLon());
+ destination.setLatitude(testOrder.getEndLat());
+
OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
- dto.setOrderId(Long.parseLong(orderId));
+ dto.setOrderId(Long.parseLong(orderId.substring(4)));
dto.setOrigin(origin);
dto.setDestination(destination);
- dto.setRouteType(2);
+ dto.setRouteType(3);
R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
if (200 != routeVoR.getCode()) {
throw new RuntimeException("获取订单线路和运费失败");
}
RouteVo routeVo = routeVoR.getData();
routeID = routeVo.getRouteID();
- //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据
- redisService.setCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + xpOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
- }else{
- RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class);
- routeID = routeVo.getRouteID();
}
}
- //驶往下车点的行程
- if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) {
- //直接取缓存中的导航数据推送给车端和乘客
- Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + xpOrder.getVehicleId());
- if(null == cacheObject){
- //没有缓存数据，重新规划线路
- //查询起点到终点的线路
- origin.setLongitude(xpOrder.getStartLon());
- origin.setLatitude(xpOrder.getStartLat());
- destination.setLongitude(xpOrder.getEndLon());
- destination.setLatitude(xpOrder.getEndLat());
- OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
- dto.setOrderId(Long.parseLong(orderId));
- dto.setOrigin(origin);
- dto.setDestination(destination);
- dto.setRouteType(2);
- R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
- if (200 != routeVoR.getCode()) {
- throw new RuntimeException("获取订单线路和运费失败");
- }
- RouteVo routeVo = routeVoR.getData();
- routeID = routeVo.getRouteID();
- //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据
- redisService.setCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + xpOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
- }else{
- RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class);
- routeID = routeVo.getRouteID();
- }
- }
- vehicleId = xpOrder.getVehicleId();
+ vehicleId = testOrder.getVehicleId();
}else{
orderType = OrderType.USER_ORDER;
//正常单
@@ -1024,10 +1124,6 @@
case ORDER_DROPOFF_COMPLETED:
//下车完成，不需要处理状态，由后续的乘客主动触发或者定时任务触发
log.info("乘客已下车，不做任何处理：{}", orderId);
-// if(!orderId.contains("XU")){
-// buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION, sequence);
-// xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.APPROVAL_REJECTION.getCode());
-// }
break;
case ORDER_COMPLETED:
//已完成
@@ -1069,17 +1165,36 @@
case ORDER_CANCELLED:
//已取消
if(!orderId.contains("XU")){
+ String taskId = orderStatusInfo.getTaskId();
+ JSONObject cancelContext = getCancelOrderContext(taskId);
+ if (shouldSkipCancelForReassign(cancelContext)) {
+ log.info("改派旧车取消回调，跳过订单取消处理，orderId: {}, taskId: {}, cancelScene: {}, context: {}",
+ orderId, taskId, cancelContext.getString("cancelScene"), cancelContext);
+ //响应消息
+ responseMessage(sequence, cancelContext.getString("vin"));
+ return;
+ }
if(orderId.contains("TEST")){
//判断订单状态，已经取消的不作处理
- Object cacheObject = redisService.getCacheObject("ORDER_STATUS:" + orderId.substring(4));
+ Object cacheObject = redisService.getCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4));
if(null != cacheObject){
int order_status = Integer.parseInt(cacheObject.toString());
if(0 == order_status){
return;
}
}
+ XpTestOrder testOrder = testOrderClient.queryOrderById(Long.valueOf(orderId.substring(4))).getData();
+ if(testOrder.getStatus() == 0 || testOrder.getStatus() >= 8){
+ return;
+ }
//行程中取消订单，按照行程结束处理
- testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
+ if(testOrder.getStatus() > 5 && testOrder.getStatus() < 8){
+ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
+ redisService.setCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), 24L, TimeUnit.HOURS);
+ }else{
+ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.CANCELED.getCode());
+ redisService.setCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.CANCELED.getCode(), 24L, TimeUnit.HOURS);
+ }
}else{
//判断订单状态，已经取消的不作处理
Object cacheObject = redisService.getCacheObject("ORDER_STATUS:" + orderId);
@@ -1089,9 +1204,20 @@
return;
}
}
- //行程中取消订单，按照行程结束处理
- buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
- xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
+ XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData();
+ if(xpOrder.getStatus() == 0 || xpOrder.getStatus() >= 8){
+ return;
+ }
+ if(xpOrder.getStatus() > 5 && xpOrder.getStatus() < 8){
+ //行程中取消订单，按照行程结束处理
+ buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode());
+ redisService.setCacheObject("ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), 24L, TimeUnit.HOURS);
+ }else{
+ buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED, sequence);
+ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.CANCELED.getCode());
+ redisService.setCacheObject("ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.CANCELED.getCode(), 24L, TimeUnit.HOURS);
+ }
}
}
break;
@@ -1132,8 +1258,16 @@
}
}
}else if(orderId.contains("TEST")){
- //测试巡游单处理逻辑
- testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.COMPLETED.getCode());
+ //测试巡游单完成状态
+ XpTestOrder testOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4))).getData();
+ if(testOrder.getOrderType() == 2){
+ XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData();
+ //响应消息
+ responseMessage(sequence, xpVehicle.getVehicleVin());
+ //测试巡游单处理逻辑
+ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.COMPLETED.getCode());
+ redisService.setCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.COMPLETED.getCode(), 24L, TimeUnit.HOURS);
+ }
}
break;
case ORDER_PAUSED:
@@ -1248,6 +1382,35 @@
}
/**
+ * 根据taskId获取订单取消的上下文信息
+ * @param taskId
+ * @return
+ */
+ private JSONObject getCancelOrderContext(String taskId) {
+ if (StringUtils.isEmpty(taskId)) {
+ return null;
+ }
+ Object cacheObject = redisService.getCacheObject(CANCEL_ORDER_CONTEXT_PREFIX + taskId);
+ if (cacheObject == null) {
+ return null;
+ }
+ return JSON.parseObject(cacheObject.toString());
+ }
+
+ /**
+ * 判断是否是改派旧车取消回调，是则跳过订单取消处理
+ * @param cancelContext
+ * @return
+ */
+ private boolean shouldSkipCancelForReassign(JSONObject cancelContext) {
+ if (cancelContext == null) {
+ return false;
+ }
+ return CancelOrderSceneEnum.REASSIGN_OLD_VEHICLE == CancelOrderSceneEnum.fromCode(
+ cancelContext.getString("cancelScene"));
+ }
+
+ /**
* 云端下发导航到XPU
* @param requestId XPU请求id
* @param routeId 高德导航id
--
Gitblit v1.7.1