From 65dcb2917a3b68e19351878e1d93f0b0ecb83cb5 Mon Sep 17 00:00:00 2001
From: 无关风月 <443237572@qq.com>
Date: 星期五, 13 二月 2026 16:49:54 +0800
Subject: [PATCH] Merge remote-tracking branch 'origin/master'
---
xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 94 ++++++++++++++++++++++++++++++++---------------
1 files changed, 64 insertions(+), 30 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 07fff2e..44f658b 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
@@ -6,6 +6,7 @@
import com.alibaba.fastjson2.JSONObject;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
import com.xiaopeng.api.dto.operation.LonLatDto;
+import com.xiaopeng.api.dto.order.PopOrderDto;
import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO;
import com.xiaopeng.api.dto.pushMessage.SendMessageDTO;
import com.xiaopeng.api.feignClient.map.XpMapClient;
@@ -35,6 +36,7 @@
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;
@@ -49,6 +51,7 @@
import javax.annotation.Resource;
import java.util.*;
+import java.util.concurrent.TimeUnit;
/**
* Kafka 消费者服务
@@ -77,6 +80,7 @@
@Resource
private XpPushMessageClient xpPushMessageClient;
+
/**
@@ -159,6 +163,17 @@
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){
+ Object sequence = redisTemplate.opsForValue().get(requestSeq);
+ if(sequence!=null){
+ String[] split = sequence.toString().split("_");
+ if(split[0].equals("pushOrder")){
+ PopOrderDto dto = new PopOrderDto();
+ dto.setOrderId(Long.valueOf(split[1]));
+ dto.setXpVehicleVin(split[2]);
+ xpOrderClient.popOrder(dto);
+ }
+ }
+
}
}
@@ -247,9 +262,6 @@
* @param msgContent
*/
public void systemStatusXpu(MsgContent msgContent) {
- //先响应消息
- Long sequence = msgContent.getHeader().getSequence();
- responseMessage(sequence);
//开始处理消息对应的业务逻辑
SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
//获取实时位姿
@@ -262,6 +274,10 @@
//获取车辆静态信息
VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
String vin = staticInfo.getVin();
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence, vin);
+ //保存车辆实时位置到redis
redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin);
//保存车辆实时信息到mongodb
@@ -314,9 +330,6 @@
* @param msgContent
*/
public void xpuNavigationRequest(MsgContent msgContent) {
- //先响应消息
- Long sequence = msgContent.getHeader().getSequence();
- responseMessage(sequence);
//开始处理消息对应的业务逻辑
NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
String orderId = naviRequest.getOrder_id();
@@ -330,15 +343,15 @@
break;
case ROUTE_TO_PICKUP:
log.info("驶往上车点的行程");
- buildNaviRequest(orderId, taskId, orderPhase);
+ buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
break;
case ROUTE_TO_DESTINATION:
log.info("驶往下车点的行程");
- buildNaviRequest(orderId, taskId, orderPhase);
+ buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
break;
case ROUTE_TO_CRUISE:
log.info("巡游阶段路线");
- buildNaviRequest(orderId, taskId, orderPhase);
+ buildNaviRequest(orderId, taskId, orderPhase, msgContent.getHeader().getSequence());
break;
}
}
@@ -352,7 +365,6 @@
public void navigationDeviationEvent(MsgContent msgContent) {
//先响应消息
Long sequence = msgContent.getHeader().getSequence();
- responseMessage(sequence);
//开始处理消息对应的业务逻辑
NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
//偏航事件类型 1：请求导航，2：取消导航
@@ -374,6 +386,8 @@
origin.setLatitude(latitude);
destination.setLongitude(116.40528);
destination.setLatitude(39.90498);
+ //响应消息
+ responseMessage(sequence, "");
} else {
//正常单
R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
@@ -381,6 +395,9 @@
throw new RuntimeException("订单不存在");
}
XpOrder xpOrder = xpOrderR.getData();
+ XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
+ //响应消息
+ responseMessage(sequence, xpVehicle.getVehicleVin());
//驶往上车点的行程
if (Arrays.asList(3).contains(xpOrder.getStatus())) {
//查询司机当前位置到起点的线路
@@ -407,7 +424,7 @@
}
RouteVo routeVo = routeVoR.getData();
//构建导航下发请求对象
- cloudOrderOperation(orderType, orderId, taskId, routeVo);
+ cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
}
}
}
@@ -420,7 +437,7 @@
* @param taskId
* @param orderPhase
*/
- public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase) {
+ public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase, Long sequence) {
//订单类型 1：乘客订单，2=巡游订单
Integer orderType = 1;
LonLatDto destination = new LonLatDto();
@@ -434,6 +451,9 @@
origin.setLatitude(39.90498);
destination.setLongitude(116.40528);
destination.setLatitude(39.90498);
+
+ //先响应消息
+ responseMessage(sequence, "");
}
} else {
//正常单
@@ -445,6 +465,9 @@
//驶往上车点的行程
if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) {
XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
+ //先响应消息
+ responseMessage(sequence, xpVehicle.getVehicleVin());
+ //查询车辆位置
VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
if (Objects.isNull(vehiclePosition)) {
throw new RuntimeException("车辆位置信息不存在");
@@ -471,9 +494,10 @@
if (200 != routeVoR.getCode()) {
throw new RuntimeException("获取订单线路和运费失败");
}
+ XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
RouteVo routeVo = routeVoR.getData();
//构建导航下发请求对象
- cloudOrderOperation(orderType, orderId, taskId, routeVo);
+ cloudOrderOperation(orderType, orderId, taskId, routeVo, xpVehicle.getVehicleVin());
}
}
@@ -483,7 +507,7 @@
*
* @param routeVo
*/
- public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo) {
+ public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo, String vin) {
//构建消息体
OrderOperation orderOperation = new OrderOperation();
orderOperation.setOrder_id(orderId);
@@ -518,20 +542,25 @@
waypoints.add(waypoint);
}
orderOperation.setWaypoints(waypoints);
+ SnowFlake snowFlake = new SnowFlake(0, 0);
+ snowFlake.withStartTimestamp(1420041600000L);
+ long sequence = snowFlake.nextId();
+ // 处理数据
+ redisTemplate.opsForValue().set(sequence,"pushOrder_"+orderId+"_"+vin,30, TimeUnit.MINUTES);
//开发发送请求到XPU
- xpuUtil.cloudOrderOperation(orderOperation);
+ xpuUtil.cloudOrderOperation(orderOperation, vin,sequence);
}
/**
* 构建响应消息
*/
- public void responseMessage(Long request_seq) {
+ public void responseMessage(Long request_seq, String vin) {
ResponseMessage responseMessage = new ResponseMessage();
responseMessage.setRequest_seq(request_seq);
responseMessage.setCode(StatusCode.OK);
responseMessage.setTimestamp(System.currentTimeMillis());
- xpuUtil.responseMessage(responseMessage);
+ xpuUtil.responseMessage(responseMessage, vin);
}
@@ -542,7 +571,6 @@
public void orderStateUpdateEvent(MsgContent msgContent) {
//先响应消息
Long sequence = msgContent.getHeader().getSequence();
- responseMessage(sequence);
//开始处理消息对应的业务逻辑
OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getContent_detail().getObject("orderStatusInfoEvent", OrderStatusInfoEvent.class);
//只处理请求消息
@@ -558,7 +586,7 @@
break;
case ORDER_RECEIVED:
//收到订单
- buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART);
+ buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART, sequence);
break;
case ORDER_STARTED:
//开始执行订单
@@ -566,7 +594,7 @@
break;
case ORDER_EN_ROUTE_TO_PICKUP:
//前往上车点
- buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP);
+ buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP, sequence);
break;
case ORDER_ARRIVED_AT_PICKUP:
//到达上车点
@@ -574,15 +602,15 @@
break;
case ORDER_WAITING_FOR_PICKUP:
//等待乘客上车
- buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS);
+ buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS, sequence);
break;
case ORDER_PASSENGER_ONBOARD:
//乘客已上车
- buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED);
+ buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence);
break;
case ORDER_EN_ROUTE_TO_DESTINATION:
//前往目的地
- buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE);
+ buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE, sequence);
break;
case ORDER_ARRIVED_AT_DESTINATION:
//到达目的地
@@ -590,11 +618,11 @@
break;
case ORDER_WAITING_FOR_DROPOFF:
//等待乘客下车
- buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS);
+ buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence);
break;
case ORDER_DROPOFF_COMPLETED:
//下车完成
- buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION);
+ buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION, sequence);
break;
case ORDER_COMPLETED:
//已完成
@@ -602,11 +630,11 @@
break;
case ORDER_CANCELLED:
//已取消
- buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED);
+ buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED, sequence);
break;
case ORDER_ARRIVED_AT_CRUISE_DESTINATION:
//到达巡游终点
- buildPushOrderStatus(orderId, null);
+ buildPushOrderStatus(orderId, null, sequence);
break;
case ORDER_PAUSED:
//订单暂停
@@ -624,11 +652,19 @@
* @param orderId
* @param status
*/
- public void buildPushOrderStatus(String orderId, OrderStatusEnum status){
+ public void buildPushOrderStatus(String orderId, OrderStatusEnum status, Long sequence){
if(orderId.contains("XU")){
//巡游单
+ //响应消息
+ responseMessage(sequence, "");
}else{
//普通单
+ R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
+ XpOrder xpOrder = xpOrderR.getData();
+ XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
+ //响应消息
+ responseMessage(sequence, xpVehicle.getVehicleVin());
+
//构建推送消息
Map<String, Object> params = new HashMap<>();
Map<String, Object> content = new HashMap<>();
@@ -638,8 +674,6 @@
params.put("content", com.alibaba.fastjson2.JSONObject.toJSONString(content));
params.put("timestamp", System.currentTimeMillis() + "");
String message = JSONObject.toJSONString(params);
- R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
- XpOrder xpOrder = xpOrderR.getData();
Long appUserId = xpOrder.getAppUserId();
SendMessageDTO dto = new SendMessageDTO();
dto.setAppUserId(appUserId + "");
--
Gitblit v1.7.1