Merge branch '1.0' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi in...

puzhibing

2026-05-12 08cc6d5cf7e93b75cfabe7dcc2aeaa9afa1117eb

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

@@ -3,6 +3,7 @@
import cn.hutool.core.lang.Snowflake;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto;
@@ -20,12 +21,14 @@
import com.xiaopeng.api.feignClient.map.XpMapClient;
import com.xiaopeng.api.feignClient.operation.ReassignOrderClient;
import com.xiaopeng.api.feignClient.operation.XpCruisePlanClient;
import com.xiaopeng.api.feignClient.operation.XpSystemSettingClient;
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.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.XpCruiseOrder;
import com.xiaopeng.api.model.order.XpOrder;
import com.xiaopeng.api.model.order.XpTestOrder;
@@ -36,6 +39,9 @@
import com.xiaopeng.api.vo.operation.RouteVo;
import com.xiaopeng.api.vo.order.OrderTrajectoryVo;
import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo;
import com.xiaopeng.automaticPilot.job.base.QuartzManager;
import com.xiaopeng.automaticPilot.job.base.TimeJobType;
import com.xiaopeng.automaticPilot.job.jobs.OrderExceptionJob;
import com.xiaopeng.automaticPilot.util.XPUUtil;
import com.xiaopeng.common.core.constant.CacheConstants;
import com.xiaopeng.common.core.constant.Constants;
@@ -43,6 +49,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.utils.DateUtils;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
import com.xiaopeng.common.redis.service.RedisService;
@@ -60,6 +67,7 @@
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
@@ -134,6 +142,9 @@

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private XpSystemSettingClient systemSettingClient;

    private static final JsonFormat.Printer jsonPrinter = JsonFormat.printer()
            .printingEnumsAsInts()
@@ -430,10 +441,12 @@
                RtTripInfo rtTripInfo = systemRuntimeStatus.getRunningOrderStatusInfo().getRtTripInfo();
                if (null != orderId && !order_id.contains("XU")) {
                    PushOrderRouteDTO dto = new PushOrderRouteDTO();
                    dto.setMessageType(PushMessageTypeEnum.ORDER_ROUTE.getDesc());
                    dto.setOrderId(orderId);
                    dto.setRouteType(1);
                    dto.setLongitude(longitude);
                    dto.setLatitude(latitude);
                    dto.setSpeed(dynamicStatus.getSpeed());
                    if (rtTripInfo.hasCurrentTask() && rtTripInfo.getCurrentTask().hasNaviProgressInfo()) {
                        dto.setDuration(rtTripInfo.getCurrentTask().getNaviProgressInfo().getRouteRemainTime());
                        dto.setDistance(rtTripInfo.getCurrentTask().getNaviProgressInfo().getRouteRemainDist());
@@ -650,7 +663,7 @@
        LonLatDto origin = new LonLatDto();
        RouteVo routeVo = null;
        String key = "";

        Float speed = 0f;

        //添加redis锁，用于在发生偏航后重新规划线路与正常推送剩余导航数据之间的冲突
        RLock lock = redissonClient.getLock("PUSH_ORDER_ROUTE_LOCK:" + orderId);
@@ -689,6 +702,13 @@
            } else if(orderId.contains("TEST")){
                R<XpTestOrder> cruiseOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4)));
                XpTestOrder testOrder = cruiseOrder.getData();

                XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData();
                VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
                if(null != vehiclePosition){
                    speed = vehiclePosition.getSpeed();
                }

                //区分测试单和测试巡游单
                if(1 == testOrder.getOrderType()){
                    orderType = OrderType.TEST_ORDER;
@@ -767,6 +787,10 @@
                }
                XpOrder xpOrder = xpOrderR.getData();
                XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
                VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class);
                if(null != vehiclePosition){
                    speed = vehiclePosition.getSpeed();
                }
                //驶往上车点的行程
                if (3 == xpOrder.getStatus()) {
                    //查询司机当前位置到起点的线路
@@ -822,6 +846,7 @@
                    dto.setRouteType(1);
                    dto.setLongitude(longitude);
                    dto.setLatitude(latitude);
                    dto.setSpeed(speed);
                    if (null != routeVo) {
                        dto.setDuration(null == routeVo.getDuration() ? 0 : routeVo.getDuration().doubleValue());
                        dto.setDistance(null == routeVo.getDistance() ? 0 : routeVo.getDistance().doubleValue());
@@ -1202,12 +1227,6 @@
                    }
                    break;
                case ORDER_PASSENGER_ONBOARD:
                    //先手动暂停触发，因为现目前车机响应状态时机不对
                    try {
                        Thread.sleep(waittime);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    //乘客已上车
                    if(!orderId.contains("XU")){
                        if(orderId.contains("TEST")){
@@ -1678,5 +1697,23 @@
        OrderPhase orderPhase = naviRequest.getOrderPhase();
        log.info("订单ID：{}", orderId);
        log.info("任务ID：{}", taskId);

        //获取系统异常订单超时配置，创建定时任务取消订单
        if(!orderId.contains("XU")){
            XpSystemSetting xpSystemSetting = systemSettingClient.getSystemSettingOne().getData();
            Map<String, ? extends Object> jobMaps =
                    new ImmutableMap.Builder<String, String>()
                            .put("id", orderId)
                            .put("orderType", orderId.contains("TEST") ? OrderType.TEST_ORDER : OrderType.USER_ORDER)
                            .build();
            QuartzManager.addJob(
                    OrderExceptionJob.class,
                    (OrderExceptionJob.name + orderId).toUpperCase(),
                    TimeJobType.ERROR_ORDER_CANCEL,
                    DateUtils.localDateTimeToDate(LocalDateTime.now().plusMinutes(xpSystemSetting.getOrderExceptionalTimeout())),
                    jobMaps
            );
        }

    }
}