添加车端消息推送重试机制

puzhibing

2026-03-06 2f8852273c3f9be8c3e01799ee6d9a4ed692ab63

添加车端消息推送重试机制

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/dto/automaticPilot/MqttPublishDto.java

@@ -1,5 +1,6 @@
package com.xiaopeng.api.dto.automaticPilot;

import com.xiaopeng.api.vo.automaticPilot.MqttPublishVo;
import lombok.Data;

@Data
@@ -16,4 +17,6 @@
     * 数据
     */
    private Object data;

    private MqttPublishVo vo;
}

 xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/OrderMissionController.java

@@ -125,7 +125,6 @@
        if (!dto.getOrderId().contains("XU") && mqttPublishDto.getCode()== 200) {
            // 修改车子为接单状态
            xpVehicleClient.updateVehicleStatus(new UpdateVehicleStateDto(dto.getVin(),1));

            return R.ok(true);
        }
        return R.ok(false);

 xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/util/XPUUtil.java

@@ -48,7 +48,6 @@
     * @param taskList
     * @return
     */
    //TODO 处理feign调用的异常
    public MqttPublishDto cloudCreateOrder(RtTaskList taskList, String vin, Long sequence) {
        MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
        MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
@@ -70,8 +69,17 @@
            log.error("Protobuf转JSON失败", e);
            throw new RuntimeException("Protobuf转JSON失败", e);
        }
        log.info("云端-->XPU 创建订单：{}", vo);
        return xpuClient.publish(vo);
        log.info("云端-->XPU 创建订单：{}", JSON.toJSONString(vo));
        try {
            MqttPublishDto dto = xpuClient.publish(vo);
            log.info("XPU返回：{}", JSON.toJSONString(dto));
            if(200 != dto.getCode()){
                dto = againPushMessage(vo);
            }
            return dto;
        }catch (Exception e){
            throw new RuntimeException("云端-->XPU 创建订单失败：", e);
        }
    }


@@ -80,7 +88,6 @@
     * @param operation
     * @return
     */
    //TODO 处理feign调用的异常
    public MqttPublishDto cloudOrderOperation(OrderOperation operation, String vin, Long sequence) {
        MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
        MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
@@ -103,9 +110,59 @@
            throw new RuntimeException("Protobuf转JSON失败", e);
        }
        log.info("云端-->XPU 修改订单：{}", vo);
        return xpuClient.publish(vo);
        try {
            MqttPublishDto dto = xpuClient.publish(vo);
            log.info("XPU返回：{}", JSON.toJSONString(dto));
            if(200 != dto.getCode()){
                dto = againPushMessage(vo);
            }
            return dto;
        }catch (Exception e){
            throw new RuntimeException("云端-->XPU 修改订单失败：", e);
        }
    }




    /**
     * 下发导航地址到车端
     *
     * @param navigationData
     * @return
     */
    public MqttPublishDto cloudPushNavigation(RestoreNavigationData navigationData, String xpuId, Long sequence) {
        MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
        MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
        headerBuilder.setMsgId(UUID.randomUUID().toString());
        headerBuilder.setTimestamp(System.currentTimeMillis());
        headerBuilder.setCmd(RTCmdType.CLOUD_NAVIGATION_RESPONSE.getValue());
        headerBuilder.setSequence(sequence);
        headerBuilder.setType(MessageType.MESSAGE_TYPE_RESPONSE);
        headerBuilder.setSource(SourceType.CLOUD);
        msgContentBuilder.setHeader(headerBuilder.build());

        msgContentBuilder.setRestoreNavigationData(navigationData);
        MqttPublishVo vo = new MqttPublishVo();
        vo.setTopic(buildTopicByXpuId(xpuId));
        try {
            vo.setPayload(jsonPrinter.print(msgContentBuilder.build()));
        } catch (Exception e) {
            log.error("Protobuf转JSON失败", e);
            throw new RuntimeException("Protobuf转JSON失败", e);
        }
        log.info("云端-->XPU 下发导航信息：{}", vo);
        try {
            MqttPublishDto dto = xpuClient.publish(vo);
            log.info("XPU返回：{}", JSON.toJSONString(dto));
            if(200 != dto.getCode()){
                dto = againPushMessage(vo);
            }
            return dto;
        }catch (Exception e){
            throw new RuntimeException("云端-->XPU 下发导航信息失败：", e);
        }
    }



@@ -139,38 +196,7 @@
        return xpuClient.publish(vo);
    }

    /**
     * 下发导航地址到车端
     *
     * @param navigationData
     * @return
     */
    //TODO 处理feign调用的异常
    public MqttPublishDto cloudPushNavigation(RestoreNavigationData navigationData, String xpuId, Long sequence) {
        MsgContent.Builder msgContentBuilder = MsgContent.newBuilder();
        MsgHeader.Builder headerBuilder = MsgHeader.newBuilder();
        headerBuilder.setMsgId(UUID.randomUUID().toString());
        headerBuilder.setTimestamp(System.currentTimeMillis());
        headerBuilder.setCmd(RTCmdType.CLOUD_NAVIGATION_RESPONSE.getValue());
        headerBuilder.setSequence(sequence);
        headerBuilder.setType(MessageType.MESSAGE_TYPE_RESPONSE);
        headerBuilder.setSource(SourceType.CLOUD);
        msgContentBuilder.setHeader(headerBuilder.build());

        msgContentBuilder.setRestoreNavigationData(navigationData);
        MqttPublishVo vo = new MqttPublishVo();
        vo.setTopic(buildTopicByXpuId(xpuId));
        try {
            vo.setPayload(jsonPrinter.print(msgContentBuilder.build()));
        } catch (Exception e) {
            log.error("Protobuf转JSON失败", e);
            throw new RuntimeException("Protobuf转JSON失败", e);
        }


        log.info("云端-->XPU 下发导航信息：{}", vo);
        return xpuClient.publish(vo);
    }


    /**
@@ -207,4 +233,26 @@
        return prefix + value + "00000000";
    }


    /**
     * 消息下发XPU重试机制
     * @param message
     * @return
     */
    public MqttPublishDto againPushMessage(MqttPublishVo message) {
        log.info("云端-->XPU 消息重试：{}", message);
        MqttPublishDto dto = null;
        for (int i = 0; i < 3; i++) {
            try {
                dto = xpuClient.publish(message);
                log.info("云端-->XPU 消息重试{}次：{}", (i+1), message);
                if (dto.getCode() == 200) {
                    break;
                }
            }catch (Exception e){
                log.error("云端-->XPU 推送失败：{}", message);
            }
        }
        return dto;
    }
}

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

@@ -52,35 +52,38 @@
    @Resource
    private RedisService redisService;

    @Autowired
    @Resource
    private XpPointService xpPointService;

    @Autowired
    @Resource
    private IXpElectronicFenceAreaService xpElectronicFenceAreaService;

    @Autowired
    @Resource
    private XpOrderClient xpOrderClient;

    @Autowired
    @Resource
    private VehiclePositionService vehiclePositionService;

    @Autowired
    @Resource
    private IXpTradingStrategyService xpTradingStrategyService;

    @Autowired
    @Resource
    private XpCruiseOrderClient xpCruiseOrderClient;

    @Autowired
    @Resource
    private XpVehicleClient xpVehicleClient;

    @Autowired
    @Resource
    private RedissonClient redissonClient;

    @Autowired
    @Resource
    private XpMapClient xpMapClient;
    //
    @Autowired
    @Resource
    private OrderMissionClient orderMissionClient;

    @Resource
    private XpElectronicFenceMapper xpElectronicFenceMapper;


    // 自定义线程池
@@ -94,9 +97,6 @@
        threadPoolTaskExecutor.setThreadNamePrefix("taskOrder-");
        threadPoolTaskExecutor.initialize();
    }

    @Autowired
    private XpElectronicFenceMapper xpElectronicFenceMapper;

    /**
     * 5秒执行一次
@@ -235,7 +235,8 @@
     */
    @SneakyThrows
    private void processPushOrder(String quoteName, int count, String uid, String[] split, String orderNo, XpTradingStrategy xpTradingStrategy, String distance, String duration, String planRule) {

        // 卡住3秒
        Thread.sleep(3000);
        System.out.println("进入推单环环节");
        if (count == 0) {
            count = 1;
@@ -256,8 +257,6 @@
        XpElectronicFence xpElectronicFence = xpElectronicFenceMapper.selectById(electronicId);

        for (int i = 0; i < count; i++) {
            // 卡住3秒
            Thread.sleep(3000);
            System.out.println("进入找车环节");
            // 根据距离找车
            // 车多则根据配置找具体车数量 根据距离最近