合并代码

puzhibing

2026-05-14 4f7fd9e879d9de211b255eb21535a034893ffbff

 xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java

@@ -7,6 +7,7 @@
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
import com.xiaopeng.api.dto.automaticPilot.PushOrderDto;
import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
import com.xiaopeng.api.dto.operation.CruisePointDto;
@@ -15,6 +16,8 @@
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.dto.vehicleConvoy.UpdateVehicleStateDto;
import com.xiaopeng.api.feignClient.automaticPilot.OrderMissionClient;
import com.xiaopeng.api.feignClient.automaticPilot.OrderMissionClient;
import com.xiaopeng.api.feignClient.automaticPilot.VehicleBuriedPointClient;
import com.xiaopeng.api.feignClient.maintenance.XpTaskClient;
@@ -25,10 +28,13 @@
import com.xiaopeng.api.feignClient.order.XpCruiseOrderClient;
import com.xiaopeng.api.feignClient.order.XpOrderClient;
import com.xiaopeng.api.feignClient.order.XpTestOrderClient;
import com.xiaopeng.api.feignClient.pushMessage.SubscribeMessageClient;
import com.xiaopeng.api.feignClient.pushMessage.WebSocketMessageClient;
import com.xiaopeng.api.feignClient.pushMessage.XpPushMessageClient;
import com.xiaopeng.api.feignClient.vehicleConvoy.UserVehicleClient;
import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
import com.xiaopeng.api.model.operation.XpSystemSetting;
import com.xiaopeng.api.model.order.OrderTrajectory;
import com.xiaopeng.api.model.order.XpCruiseOrder;
import com.xiaopeng.api.model.order.XpOrder;
import com.xiaopeng.api.model.order.XpTestOrder;
@@ -38,6 +44,7 @@
import com.xiaopeng.api.vo.automaticPilot.protobuf.*;
import com.xiaopeng.api.vo.operation.RouteVo;
import com.xiaopeng.api.vo.order.OrderTrajectoryVo;
import com.xiaopeng.api.vo.pushMessage.UpdateEndAddStatusVo;
import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo;
import com.xiaopeng.automaticPilot.job.base.QuartzManager;
import com.xiaopeng.automaticPilot.job.base.TimeJobType;
@@ -49,6 +56,7 @@
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.exception.ServiceException;
import com.xiaopeng.common.core.utils.DateUtils;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
@@ -91,6 +99,12 @@
    private XpOrderClient xpOrderClient;

    @Resource
    private OrderMissionClient orderMissionClient;

    @Resource
    private XpCruiseOrderClient xpCruiseOrderClient;

    @Resource
    private MongoTemplate mongoTemplate;

    @Resource
@@ -110,9 +124,6 @@

    @Resource
    private UserVehicleClient userVehicleClient;

    @Resource
    private XpCruiseOrderClient xpCruiseOrderClient;

    @Resource
    private RedisService redisService;
@@ -142,6 +153,9 @@

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private WebSocketMessageClient webSocketMessageClient;

    @Resource
    private XpSystemSettingClient systemSettingClient;
@@ -320,7 +334,22 @@
                        }
                        dto.setXpVehicleVin(vin);
                        dto.setOrderType(orderType);
                        xpOrderClient.popOrder(dto);
                        R<?> r = xpOrderClient.popOrder(dto);
                        if(200 != r.getCode()){
                            //业务执行失败，给差车端发送取消任务报文，继续执行巡游任务
                            PushOrderDto pushOrderDto = new PushOrderDto();
                            pushOrderDto.setOrderId(orderId);
                            pushOrderDto.setVin(vin);
                            pushOrderDto.setOperation_type(OrderOperationType.CLOUD_CANCEL_ORDER);
                            orderMissionClient.cancelOrder(pushOrderDto);
                            log.info("用户取消订单成功，开始执行巡游订单 {}", orderId);
                            //车辆继续巡游
                            xpOrderClient.addNewCruiseMission(orderId);
                            log.error("处理车辆接单后的业务逻辑异常，车辆继续执行巡游 {}", orderId);
                            //放入队列，重新执行派单
                            xpOrderClient.againPushOrder(orderId);
                            return;
                        }
                        log.info("订单状态更新：{}", response.getResults(0).getOrderId());
                    }
                } catch (InvalidProtocolBufferException e) {
@@ -462,6 +491,8 @@
                    orderTrajectory.setCreateTime(new Date());
                    //只有乘客订单需要推送剩余导航数据
                    if(!order_id.contains("TEST")){
                        //推送修改目的地权限状态
                        pushUpdateEndAddressStatus(orderId);
                        xpPushMessageClient.pushOrderRoute(dto);//存储历史轨迹
                        // 给连环派单的用户推送
                        String json = redisService.getCacheObject(CacheConstants.nextOrder(order_id));
@@ -486,7 +517,65 @@
    }


    /**
     * 推送修改目的地权限状态
     * @param orderId
     */
    private void pushUpdateEndAddressStatus(Long orderId){
        XpOrder xpOrder = xpOrderClient.queryOrderById(orderId).getData();
        Boolean hasKey = redisService.hasKey("UPDATE_END_ADDRESS_STATUS_MSG:" + xpOrder.getAppUserId());
        if(hasKey){
            return;
        }
        Integer status = 0;
        if(xpOrder.getStatus() < 6){
            status = 1;
        }
        if(xpOrder.getStatus() == 6){
            XpSystemSetting xpSystemSetting = systemSettingClient.getSystemSettingOne().getData();
            //行程中剩余里程
            Integer xczSyDistance = xpSystemSetting.getXczSyDistance();
            //行程中剩余时间
            Integer xczSyTime = xpSystemSetting.getXczSyTime();
            //行程中修改目地次数
            Integer xczCaCount = xpSystemSetting.getXczCaCount();
            //行程中豁免时速 km/h
            Integer xczHmSpeedPerHour = xpSystemSetting.getXczHmSpeedPerHour();
            //行程中豁免持续时间 分钟
            Integer xczHmDuration = xpSystemSetting.getXczHmDuration();
            Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId());
            if(null != cacheObject){
                RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class);
                //剩余时间 秒
                Integer duration = routeVo.getDuration();
                //剩余距离 米
                Integer distance = routeVo.getDistance();
                if(xczSyDistance < distance && xczSyTime * 60 < duration && xpOrder.getOrderChange() < xczCaCount){
                    status = 1;
                }else{
                    //计算豁免
                    //计算平均时速小于设定值，可豁免限制
                    LocalDateTime localDateTime = LocalDateTime.now().minusMinutes(xczHmDuration);
                    List<OrderTrajectory> orderTrajectories = mongoTemplate.find(Query.query(Criteria.where("orderId").is(orderId).gt("createTime").is(localDateTime)), OrderTrajectory.class);
                    double speed = orderTrajectories.stream().map(OrderTrajectory::getSpeed).reduce(Double::sum).orElse(0D) / orderTrajectories.size();
                    if(speed <= xczHmSpeedPerHour){
                        status = 1;
                    }
                }
            }
        }
        UpdateEndAddStatusVo vo = new UpdateEndAddStatusVo();
        vo.setAppUserId(xpOrder.getAppUserId());
        JSONObject message = new JSONObject();
        message.put("type", "UPDATE_END_ADDRESS_STATUS");
        JSONObject content = new JSONObject();
        content.put("status", status);
        message.put("content", content);
        message.put("timestamp", System.currentTimeMillis());
        vo.setMessage(message.toString());
        webSocketMessageClient.sendUpdateEndAddStatus(vo);

    }

    /**
     * XPU发布系统运行状态---车辆实时位置
@@ -1160,7 +1249,8 @@

    }

    @Resource OrderMissionClient orderMissionClient;


    /**
     * 订单状态更新事件
     * @param msgContent
@@ -1570,6 +1660,7 @@
        return JSON.parseObject(cacheObject.toString());
    }


    private JSONObject resolveReassignCancelContext(String orderId, OrderStatusInfo orderStatusInfo) {
        if (OrderStateChangeSource.ORDER_STATE_CHANGE_SOURCE_VEHICLE != orderStatusInfo.getChangeSource()) {
            return null;