[feat]连环派代-合并路线和新订单的流转与推送

xuhy

2026-04-16 7bddec9926d953e9362c2a32f1c4a150b9d8c5d2

[feat]连环派代-合并路线和新订单的流转与推送

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

@@ -13,11 +13,10 @@
import com.xiaopeng.api.dto.operation.CruisePointDto;
import com.xiaopeng.api.dto.operation.LonLatDto;
import com.xiaopeng.api.dto.operation.ReassignOrderDTO;
import com.xiaopeng.api.dto.order.AddCruiseMissionDto;
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.VehicleBuriedPointClient;
import com.xiaopeng.api.feignClient.maintenance.XpTaskClient;
import com.xiaopeng.api.feignClient.map.XpMapClient;
@@ -63,6 +62,7 @@
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;


/**
@@ -417,8 +417,8 @@
                mongoTemplate.updateFirst(new Query(Criteria.where("vin").is(vin)), update, VehiclePosition.class);
            }
            //调用推送服务接口，给乘客推送剩余轨迹和里程时长
            RtTripInfo rtTripInfo = systemRuntimeStatus.getRunningOrderStatusInfo().getRtTripInfo();
            if (null != orderId && !order_id.contains("XU")) {
                RtTripInfo rtTripInfo = systemRuntimeStatus.getRunningOrderStatusInfo().getRtTripInfo();
                PushOrderRouteDTO dto = new PushOrderRouteDTO();
                dto.setOrderId(orderId);
                dto.setRouteType(1);
@@ -440,6 +440,12 @@
                //只有乘客订单需要推送剩余导航数据
                if(!order_id.contains("TEST")){
                    xpPushMessageClient.pushOrderRoute(dto);//存储历史轨迹
                    // 给连环派单的用户推送
                    Long nextOrderId = redisService.getCacheObject(CacheConstants.nextOrder(order_id));
                    if (null != nextOrderId){
                        dto.setOrderId(orderId);
                        xpPushMessageClient.pushOrderRoute(dto);
                    }
                    xpOrderClient.saveVehicleLocation(orderTrajectory);
                }else{
                    testOrderClient.saveVehicleLocation(orderTrajectory);
@@ -1100,8 +1106,8 @@

    }

    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(20);

    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(40);
    @Resource OrderMissionClient orderMissionClient;
    /**
     * 订单状态更新事件
     * @param msgContent
@@ -1196,16 +1202,23 @@
                    log.info("到达目的地，不做任何处理：{}", orderId);
                    break;
                case ORDER_WAITING_FOR_DROPOFF:
                    // 根据车辆id查询订单，如果未完成订单数量不唯一则下发循环订单任务
                    List ret = new ArrayList(2);//TODO
                    if(ret.size() > 1){
                    Long id = redisService.getCacheObject(CacheConstants.nextOrder(orderId));
                    if(id != null){
                        scheduler.schedule(() -> {
                            //下发循环订单任务
//                        xpOrderClient.createLoopOrderTask(vin, orderId);
                            log.info("下发循环订单任务：{}", orderId);
                            // 修改订单状态，和推送消息的状态
                        }, 3, TimeUnit.SECONDS);

                            XpOrder xpOrder = xpOrderClient.queryOrderById(id).getData();
                            XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
                            PushOrderDto pushOrderDto = new PushOrderDto();
                            pushOrderDto.setOrderId(xpOrder.getId().toString());
                            pushOrderDto.setStartLatitude(xpOrder.getStartLat());
                            pushOrderDto.setStartLongitude(xpOrder.getStartLon());
                            pushOrderDto.setEndLatitude(xpOrder.getEndLat());
                            pushOrderDto.setEndLongitude(xpOrder.getEndLon());
                            pushOrderDto.setEndPoiName(xpOrder.getEndPointName());
                            pushOrderDto.setOrderType(OrderType.USER_ORDER);
                            pushOrderDto.setStage(1);
                            pushOrderDto.setVin(xpVehicle.getVehicleVin());
                            orderMissionClient.pushOrder(pushOrderDto);
                        }, 3, TimeUnit.MINUTES);
                        break;
                    }


 xp-robotaxi-common/xp-common-core/src/main/java/com/xiaopeng/common/core/constant/CacheConstants.java

@@ -153,5 +153,22 @@
     */
    public static final String reassign_Order = "reassign_order";

    static final String ORDER_NAVIGATION_END_DATA = "ORDER_NAVIGATION_END_DATA:";

    public static String navigationEnd(String key) {
        return ORDER_NAVIGATION_END_DATA + key;
    }

    static final String ORDER_NAVIGATION_START_DATA = "ORDER_NAVIGATION_START_DATA:";

    public static String navigationStart(String key) {
        return ORDER_NAVIGATION_START_DATA + key;
    }

    static final String NEXT_ORDER = "NEXT_ORDER:";

    public static String nextOrder(String currentOrder) {
        return NEXT_ORDER + currentOrder;
    }

}

 xp-robotaxi-operation-boot/src/main/java/com/xiaopeng/operation/utils/TaskOrderUtil.java

@@ -49,7 +49,9 @@

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@@ -107,6 +109,8 @@

    // 自定义线程池
    private static final ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();

//    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(20);

    static {
        threadPoolTaskExecutor.setCorePoolSize(20);
@@ -1059,16 +1063,8 @@
                            continue;
                        }
                        log.info("关闭巡游单成功 {} {}", vehicleId1.getPlateNumber(), orderNo);
                        PushOrderDto pushOrderDto = new PushOrderDto();
                        PushOrderDto pushOrderDto = buildPushOrderDto(xpOrder);
                        pushOrderDto.setVin(vehiclePosition.getVin());
                        pushOrderDto.setOrderId(xpOrder.getId().toString());
                        pushOrderDto.setStartLatitude(xpOrder.getStartLat());
                        pushOrderDto.setStartLongitude(xpOrder.getStartLon());
                        pushOrderDto.setEndLatitude(xpOrder.getEndLat());
                        pushOrderDto.setEndLongitude(xpOrder.getEndLon());
                        pushOrderDto.setEndPoiName(xpOrder.getEndPointName());
                        pushOrderDto.setOrderType(OrderType.USER_ORDER);
                        pushOrderDto.setStage(1);
                        // 调用推送服务  orderId + carVin
                        orderMissionClient.pushOrder(pushOrderDto);
                        log.info("开始推送 {} 订单到XPU", xpOrder.getOrderNumber());
@@ -1090,9 +1086,8 @@
            if (vehiclePositionVos.getCode() != 200) {
                continue;
            }

            LonLatDto lonLatDto = new LonLatDto(Double.parseDouble(split[0]), Double.parseDouble(split[1]));
            RouteAndFareDto routeAndFareDto = new RouteAndFareDto();
            routeAndFareDto.setDestination(lonLatDto);

            Optional<XpOrder> lucyOrderOpt = servingVehicles.getData().stream()
                    //获取位置
@@ -1101,37 +1096,51 @@
                    .map(xpOrderClient::queryOrderById)
                    .filter(R::isSuccess)
                    .map(R::getData)
                    .filter(o -> isClose(o, routeAndFareDto))
                    .filter(o -> {
                        RouteAndFareDto routeAndFareDto = new RouteAndFareDto();
                        routeAndFareDto.setDestination(lonLatDto);
                        return isClose(o, routeAndFareDto);
                    })
                    .findAny();

            if (lucyOrderOpt.isPresent()) {
                XpOrder lucyOder = lucyOrderOpt.get();
                log.info("有空闲车辆 {} 订单 {}", lucyOder.getOrderNumber(), orderNo);
                XpVehicle xpVehicle = xpVehicleClient.queryById(lucyOder.getVehicleId()).getData();
                // 设置路线信息
                RouteAndFareDto routeAndFareDto = new RouteAndFareDto();
                routeAndFareDto.setDestination(lonLatDto);
                routeAndFareDto.setOrigin(new LonLatDto(lucyOder.getEndLon(),lucyOder.getEndLat()));
                routeAndFareDto.setOrigin(lonLatDto);

                RouteAndFareDto routeAndFare = null;//TODO 获取导航路线
                if (null == routeAndFare) { //轨迹不在运营区
                    continue;
                }
                // 当轨迹是否在运营区内，存储
                xpOrder.setStatus(1);
                xpOrderClient.updateOrder(xpOrder);
                scheduler.scheduleWithFixedDelay(this::pushProgress, 0, 5, TimeUnit.SECONDS);
                Object cacheObject = redisService.getCacheObject(CacheConstants.navigationEnd(lucyOder.getAppUserId().toString()));
                RouteVo routeRemain = JSON.parseObject(cacheObject.toString(), RouteVo.class);
                RouteVo routeToOrigin = xpMapClient.getDistanceAndDuration(routeAndFareDto).getData();
                routeToOrigin.getPolyline().addAll(0, routeRemain.getPolyline());
                routeToOrigin.setDuration(routeToOrigin.getDuration() + routeRemain.getDuration());
                routeToOrigin.setDistance(routeToOrigin.getDistance() + routeRemain.getDistance());
                PushOrderDto pushOrderDto = buildPushOrderDto(xpOrder);
                pushOrderDto.setVin(xpVehicle.getVehicleVin());
                // 订单流转
                orderMissionClient.pushOrder(pushOrderDto);
                redisService.setCacheObject(CacheConstants.nextOrder(lucyOder.getId().toString()), xpOrder.getId());
                redisService.setCacheObject(CacheConstants.navigationStart(xpOrder.getAppUserId().toString()), routeToOrigin);
                redisService.removeByPattern(quoteName,xpOrder.getOrderNumber());
                log.info("将订单 {} 从队列 {} 弹出去", orderNo, quoteName);
            }
        }
        return false;
    }

    private void pushProgress() {
        //TODO 调用push message 服务
    private static PushOrderDto buildPushOrderDto(XpOrder xpOrder) {
        PushOrderDto pushOrderDto = new PushOrderDto();
//        pushOrderDto.setVin(lucyOrder.get().getVin());
//        pushOrderDto.setOrderId(xpOrder.getId().toString());
//        pushOrderDto.setStartLatitude(xpOrder.getStartLat());
//        pushOrderDto.setStartLongitude(xpOrder.getStartLon());
//        pushOrderDto.setEndLatitude(xpOrder.getEndLat());
//        pushOrderDto.setEndLongitude(xpOrder.getEndLon());
//        pushOrderDto.setOrderType(OrderType.USER_ORDER);
        pushOrderDto.setOrderId(xpOrder.getId().toString());
        pushOrderDto.setStartLatitude(xpOrder.getStartLat());
        pushOrderDto.setStartLongitude(xpOrder.getStartLon());
        pushOrderDto.setEndLatitude(xpOrder.getEndLat());
        pushOrderDto.setEndLongitude(xpOrder.getEndLon());
        pushOrderDto.setEndPoiName(xpOrder.getEndPointName());
        pushOrderDto.setOrderType(OrderType.USER_ORDER);
        pushOrderDto.setStage(1);
        return pushOrderDto;
    }

    private boolean isClose(XpOrder xpOrder,RouteAndFareDto routeAndFareDto) {
@@ -1142,11 +1151,6 @@

    @Resource
    private MongoTemplate mongoTemplate;
    boolean isClose(Long orderId) {
        // yong id 去查目的地坐标，判断是否小于800
        return false;
    }



    /**