From 8712e98a272ef5d1db34abfc8387bbaf04953df8 Mon Sep 17 00:00:00 2001
From: puzhibing <393733352@qq.com>
Date: 星期五, 13 二月 2026 00:23:00 +0800
Subject: [PATCH] 合并代码
---
xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java | 272 ++++++++++++++++++++++++++++++++++++++++++++++++------
1 files changed, 242 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 eb3c28b..1f46d8e 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
@@ -2,8 +2,8 @@
import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.util.IdUtil;
-import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
import com.xiaopeng.api.dto.operation.LonLatDto;
import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO;
@@ -16,17 +16,25 @@
import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
import com.xiaopeng.api.vo.automaticPilot.VehiclePosition;
import com.xiaopeng.api.vo.automaticPilot.proto.down.GeoPoint;
+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.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.operation.RouteVo;
import com.xiaopeng.automaticPilot.util.XPUUtil;
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 com.xiaopeng.common.redis.service.RedisService;
@@ -40,9 +48,7 @@
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
+import java.util.*;
/**
* Kafka 消费者服务
@@ -88,6 +94,7 @@
switch (cmd) {
case SYSTEM_STATUS_XPU:
log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
+ systemStatusXpu(msgContent);
break;
case XPU_NAVIGATION_REQUEST:
log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
@@ -110,6 +117,7 @@
break;
case ORDER_STATE_UPDATE_EVENT:
log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ orderStateUpdateEvent(msgContent);
break;
case POINT_ARRIVAL_STATUS_EVENT:
log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
@@ -122,6 +130,7 @@
break;
case NAVIGATION_DEVIATION_EVENT:
log.info("导航偏航(XPU→云端)：{}", message);
+ navigationDeviationEvent(msgContent);
break;
case EMERGENCY_STOP_EVENT:
log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
@@ -141,7 +150,6 @@
default:
log.info("消息不在处理范围内，忽略：{}", message);
}
-
}
@@ -185,6 +193,7 @@
break;
case ORDER_STATE_UPDATE_EVENT:
log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
+ orderStateUpdateEvent(msgContent);
break;
case POINT_ARRIVAL_STATUS_EVENT:
log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
@@ -197,6 +206,7 @@
break;
case NAVIGATION_DEVIATION_EVENT:
log.info("导航偏航(XPU→云端)：{}", message);
+ navigationDeviationEvent(msgContent);
break;
case EMERGENCY_STOP_EVENT:
log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
@@ -225,9 +235,14 @@
/**
* XPU发布系统运行状态---车辆实时位置
+ *
* @param msgContent
*/
- public void systemStatusXpu(MsgContent msgContent){
+ public void systemStatusXpu(MsgContent msgContent) {
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence);
+ //开始处理消息对应的业务逻辑
SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
//获取实时位姿
RealtimePose pose = systemRuntimeStatus.getPose();
@@ -243,7 +258,7 @@
//保存车辆实时信息到mongodb
VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class);
- if(null == vehiclePosition){
+ if (null == vehiclePosition) {
vehiclePosition = new VehiclePosition();
}
vehiclePosition.setVin(vin);
@@ -260,18 +275,18 @@
Long orderId = null;
String order_id = runningOrderStatusInfo.getOrder_id();
vehiclePosition.setHasOrder(0);
- if(StringUtils.isNotEmpty(order_id)){
- if(order_id.contains("XU")){
+ if (StringUtils.isNotEmpty(order_id)) {
+ if (order_id.contains("XU")) {
vehiclePosition.setHasOrder(1);
orderId = Long.parseLong(order_id.substring(2));
- }else{
+ } else {
orderId = Long.parseLong(order_id);
}
}
vehiclePosition.setOrderId(orderId);
mongoTemplate.save(vehiclePosition);
//调用推送服务接口
- if(null != orderId){
+ if (null != orderId) {
RtTripInfo rtTripInfo = systemRuntimeStatus.getRt_trip_info();
PushOrderRouteDTO dto = new PushOrderRouteDTO();
dto.setOrderId(orderId);
@@ -285,21 +300,23 @@
}
-
-
-
/**
* XPU向云端请求导航
+ *
* @param msgContent
*/
- public void xpuNavigationRequest(MsgContent 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();
String taskId = naviRequest.getTask_id();
OrderPhase orderPhase = naviRequest.getOrder_phase();
log.info("订单ID：{}", orderId);
log.info("任务ID：{}", taskId);
- switch (orderPhase){
+ switch (orderPhase) {
case IDLE:
log.info("订单的任务未开始执行");
break;
@@ -309,47 +326,119 @@
break;
case ROUTE_TO_DESTINATION:
log.info("驶往下车点的行程");
+ buildNaviRequest(orderId, taskId, orderPhase);
break;
case ROUTE_TO_CRUISE:
log.info("巡游阶段路线");
+ buildNaviRequest(orderId, taskId, orderPhase);
break;
}
}
/**
+ * 车辆偏航后重新规划线路
+ *
+ * @param msgContent
+ */
+ public void navigationDeviationEvent(MsgContent msgContent) {
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence);
+ //开始处理消息对应的业务逻辑
+ NavigationDeviationEvent deviationEvent = msgContent.getContent_detail().getObject("deviationEvent", NavigationDeviationEvent.class);
+ //偏航事件类型 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();
+ //订单类型 1：乘客订单，2=巡游订单
+ Integer orderType = 1;
+ LonLatDto destination = new LonLatDto();
+ LonLatDto origin = new LonLatDto();
+ if (orderId.contains("XU")) {
+ orderType = 2;
+ // todo 下发巡游单的导航数据
+ //巡游单
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(116.40528);
+ destination.setLatitude(39.90498);
+ } else {
+ //正常单
+ R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
+ if (200 != xpOrderR.getCode()) {
+ throw new RuntimeException("订单不存在");
+ }
+ XpOrder xpOrder = xpOrderR.getData();
+ //驶往上车点的行程
+ if (Arrays.asList(3).contains(xpOrder.getStatus())) {
+ //查询司机当前位置到起点的线路
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(xpOrder.getStartLon());
+ destination.setLatitude(xpOrder.getStartLat());
+ }
+ //驶往下车点的行程
+ if (Arrays.asList(6).contains(xpOrder.getStatus())) {
+ //查询起点到终点的线路
+ origin.setLongitude(longitude);
+ origin.setLatitude(latitude);
+ destination.setLongitude(xpOrder.getEndLon());
+ destination.setLatitude(xpOrder.getEndLat());
+ }
+ OrderRouteAndFareDto dto = new OrderRouteAndFareDto();
+ dto.setOrderId(Long.parseLong(orderId));
+ dto.setOrigin(origin);
+ dto.setDestination(destination);
+ R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
+ if (200 != routeVoR.getCode()) {
+ throw new RuntimeException("获取订单线路和运费失败");
+ }
+ RouteVo routeVo = routeVoR.getData();
+ //构建导航下发请求对象
+ cloudOrderOperation(orderType, orderId, taskId, routeVo);
+ }
+ }
+ }
+
+
+ /**
* 构建导航请求
+ *
* @param orderId
* @param taskId
* @param orderPhase
*/
- public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase){
+ public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase) {
//订单类型 1：乘客订单，2=巡游订单
Integer orderType = 1;
LonLatDto destination = new LonLatDto();
LonLatDto origin = new LonLatDto();
- if(orderId.contains("XU")){
+ if (orderId.contains("XU")) {
orderType = 2;
// todo 下发巡游单的导航数据
//巡游单
- if(orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)){
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) {
origin.setLongitude(116.40528);
origin.setLatitude(39.90498);
destination.setLongitude(116.40528);
destination.setLatitude(39.90498);
}
- }else{
+ } else {
//正常单
R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
- if(200 != xpOrderR.getCode()){
+ if (200 != xpOrderR.getCode()) {
throw new RuntimeException("订单不存在");
}
XpOrder xpOrder = xpOrderR.getData();
//驶往上车点的行程
- if(orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)){
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) {
XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
- if(Objects.isNull(vehiclePosition)){
+ if (Objects.isNull(vehiclePosition)) {
throw new RuntimeException("车辆位置信息不存在");
}
//查询司机当前位置到起点的线路
@@ -359,7 +448,7 @@
destination.setLatitude(xpOrder.getStartLat());
}
//驶往下车点的行程
- if(orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)){
+ if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) {
//查询起点到终点的线路
origin.setLongitude(xpOrder.getStartLon());
origin.setLatitude(xpOrder.getStartLat());
@@ -371,7 +460,7 @@
dto.setOrigin(origin);
dto.setDestination(destination);
R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
- if(200 != routeVoR.getCode()){
+ if (200 != routeVoR.getCode()) {
throw new RuntimeException("获取订单线路和运费失败");
}
RouteVo routeVo = routeVoR.getData();
@@ -383,9 +472,10 @@
/**
* 云端订单操作----下发导航数据
+ *
* @param routeVo
*/
- public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo){
+ public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo) {
//构建消息体
OrderOperation orderOperation = new OrderOperation();
orderOperation.setOrder_id(orderId);
@@ -399,14 +489,14 @@
LonLatDto lonLatDto = polyline.get(i);
Waypoint waypoint = new Waypoint();
waypoint.setWaypoint_id(UUID.randomUUID().toString());
- if(0 == i){
+ if (0 == i) {
waypoint.setPoint_type(PointType.START_POINT);
//订单类型不同，点位的动作类型不同
waypoint.setAction_type(1 == orderType ? ActionType.PICK_UP : ActionType.ACTION_UNSPECIFIED);
- }else if(polyline.size() - 1 == i){
+ } else if (polyline.size() - 1 == i) {
waypoint.setPoint_type(PointType.END_POINT);
waypoint.setAction_type(1 == orderType ? ActionType.DROP_OFF : ActionType.ACTION_UNSPECIFIED);
- }else{
+ } else {
waypoint.setPoint_type(PointType.WAY_POINT);
//订单类型不同，点位的动作类型不同
waypoint.setAction_type(ActionType.ACTION_UNSPECIFIED);
@@ -425,10 +515,132 @@
}
- public void c(MsgContent msgContent){
+ /**
+ * 构建响应消息
+ */
+ public void responseMessage(Long request_seq) {
+ ResponseMessage responseMessage = new ResponseMessage();
+ responseMessage.setRequest_seq(request_seq);
+ responseMessage.setCode(StatusCode.OK);
+ responseMessage.setTimestamp(System.currentTimeMillis());
+ xpuUtil.responseMessage(responseMessage);
+ }
+ /**
+ * 订单状态更新事件
+ * @param msgContent
+ */
+ public void orderStateUpdateEvent(MsgContent msgContent) {
+ //先响应消息
+ Long sequence = msgContent.getHeader().getSequence();
+ responseMessage(sequence);
+ //开始处理消息对应的业务逻辑
+ OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getContent_detail().getObject("orderStatusInfoEvent", OrderStatusInfoEvent.class);
+ //只处理请求消息
+ 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();
+ //普通单
+ com.xiaopeng.api.vo.automaticPilot.proto.up.orderStateUpdateEvent.OrderStatus status = orderStatusInfo.getStatus();
+ switch (status) {
+ case ORDER_STATUS_UNSPECIFIED:
+ //订单状态未知
+ log.info("订单状态未知，不做任何处理：{}", orderId);
+ break;
+ case ORDER_RECEIVED:
+ //收到订单
+ buildPushOrderStatus(orderId, OrderStatusEnum.READY_DEPART);
+ break;
+ case ORDER_STARTED:
+ //开始执行订单
+ log.info("开始执行订单，不做任何处理：{}", orderId);
+ break;
+ case ORDER_EN_ROUTE_TO_PICKUP:
+ //前往上车点
+ buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP);
+ break;
+ case ORDER_ARRIVED_AT_PICKUP:
+ //到达上车点
+ log.info("到达上车点，不做任何处理：{}", orderId);
+ break;
+ case ORDER_WAITING_FOR_PICKUP:
+ //等待乘客上车
+ buildPushOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS);
+ break;
+ case ORDER_PASSENGER_ONBOARD:
+ //乘客已上车
+ buildPushOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED);
+ break;
+ case ORDER_EN_ROUTE_TO_DESTINATION:
+ //前往目的地
+ buildPushOrderStatus(orderId, OrderStatusEnum.IN_SERVICE);
+ break;
+ case ORDER_ARRIVED_AT_DESTINATION:
+ //到达目的地
+ log.info("到达目的地，不做任何处理：{}", orderId);
+ break;
+ case ORDER_WAITING_FOR_DROPOFF:
+ //等待乘客下车
+ buildPushOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS);
+ break;
+ case ORDER_DROPOFF_COMPLETED:
+ //下车完成
+ buildPushOrderStatus(orderId, OrderStatusEnum.APPROVAL_REJECTION);
+ break;
+ case ORDER_COMPLETED:
+ //已完成
+ log.info("已完成，不做任何处理：{}", orderId);
+ break;
+ case ORDER_CANCELLED:
+ //已取消
+ buildPushOrderStatus(orderId, OrderStatusEnum.CANCELED);
+ break;
+ case ORDER_ARRIVED_AT_CRUISE_DESTINATION:
+ //到达巡游终点
+ buildPushOrderStatus(orderId, null);
+ break;
+ case ORDER_PAUSED:
+ //订单暂停
+ log.info("订单暂停，不做任何处理：{}", orderId);
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+ /**
+ * 构建订单状态推送消息
+ * @param orderId
+ * @param status
+ */
+ public void buildPushOrderStatus(String orderId, OrderStatusEnum status){
+ if(orderId.contains("XU")){
+ //巡游单
+ }else{
+ //普通单
+ //构建推送消息
+ Map<String, Object> params = new HashMap<>();
+ Map<String, Object> content = new HashMap<>();
+ 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("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 + "");
+ dto.setMessage(message);
+ R<String> stringR = xpPushMessageClient.sendStatus(dto);
+ if (R.isError(stringR)) {
+ log.error("推送消息失败：{}", stringR.getMsg());
+ }
+ }
}
}
--
Gitblit v1.7.1