Merge branch 'master' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi

liujie

2026-04-01 9adef3dd7723930d44f72713c3cf28e50b002055

 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