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

liujie

2026-02-12 2064bc46d0d716a3a6eb1ab9c113ee8b7748a613

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

 .gitignore

@@ -15,7 +15,7 @@

target

#*.yaml
#*.yml
*.yaml
*.yml
*.log.DS_Store
**/.DS_Store

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/dto/pushMessage/PushOrderRouteDTO.java

@@ -31,12 +31,8 @@
    @ApiModelProperty(value = "剩余时间")
    private Integer duration;

    @NotNull(message = "用户id不能为空")
    @ApiModelProperty(value = "用户id")
    private Long appUserId;

    @NotNull(message = "轨迹类型不能为空")
    @ApiModelProperty(value = "轨迹类型 1=接单轨迹 2=服务轨迹 3=巡游轨迹")
    @NotNull(message = "订单类型不能为空")
    @ApiModelProperty(value = "订单类型 1=实时订单 2=巡游订单")
    private Integer routeType;

}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/factory/vehicleConvoy/VehicleControlClientFallbackFactory.java

@@ -13,7 +13,7 @@
    public VehicleControlClient create(Throwable cause) {
        return new VehicleControlClient() {
            @Override
            public BaseImOpenDto queryVehicleTemperature(VehicleTemperatureVo vo, Long uid) {
            public BaseImOpenDto queryVehicleTemperature(VehicleTemperatureVo vo, String uid) {
                log.error("查询车辆内温度失败：{}", cause.getMessage());
                return null;
            }

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/feignClient/vehicleConvoy/VehicleControlClient.java

@@ -24,7 +24,7 @@
     * @return
     */
    @GetMapping("/logan/api/v1/vehicles/state")
    BaseImOpenDto queryVehicleTemperature(VehicleTemperatureVo vo, @RequestHeader("uid") Long uid);
    BaseImOpenDto queryVehicleTemperature(VehicleTemperatureVo vo, @RequestHeader("uid") String uid);


    /**

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/VehiclePosition.java

@@ -22,13 +22,9 @@
)
public class  VehiclePosition implements Serializable {
    /**
     * 车辆id
     */
    @Id
    private Long vehicle_id;
    /**
     * 车辆VIN
     */
    @Id
    private String vin;
    /**
     * 车辆所属围栏id
@@ -39,19 +35,21 @@
     */
    private GeoJsonPoint point;
    /**
     * 运营状态 1运营中 2维修中 3已收车 4车辆异常
     * 航向，单位：度，北零顺时针。范围：[0-360)。0 => 北, 90 => 东, 180 => 南, 270 => 西
     */
    private Integer status;

    private Double course;
    /**
     * 车辆状态 1接单中 0空闲  巡游单不做处理 还是空闲
     * 海拔 单位米
     */
    private Integer state;

    private Float alt;
    /**
     * 当前速度
     */
    private Float speed;
    /**
     * 当前电量
     */
    private Integer currentElectricity;
    private Float currentElectricity;

    /**
     * 是否有巡游单子 1有 0否
@@ -62,9 +60,4 @@
     * 订单id
     */
    private Long orderId;





}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/ActionType.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/OrderOperation.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/OrderOperationState.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/OrderOperationStatus.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/OrderOperationType.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/OrderType.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/PointType.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/TargetPoint.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/down/cloudPushAddress/Waypoint.java

File was deleted

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/vo/automaticPilot/proto/up/systemStatusXpu/VehicleDynamicStatus.java

@@ -23,4 +23,20 @@
     * 时间戳(毫秒)
     */
    private Long timestamp;
    /**
     * 油门开度(0-100%)
     */
    private Float throttle;
    /**
     * 刹车压力(0-100%)
     */
    private Float brake;
    /**
     * 手刹状态
     */
    private Boolean parking_brake;
    /**
     * 电池电量(0-100%)
     */
    private Float battery_level;
}

 xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/controller/KafkaTestController.java

@@ -1,11 +1,12 @@
package com.xiaopeng.automaticPilot.controller;

import com.xiaopeng.automaticPilot.service.KafkaProducerService;
import com.alibaba.fastjson2.JSON;
import com.xiaopeng.api.vo.automaticPilot.proto.up.MsgContent;
import com.xiaopeng.automaticPilot.service.KafkaConsumerService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.*;

/**
@@ -18,138 +19,40 @@
public class KafkaTestController {

    @Autowired
    private KafkaProducerService kafkaProducerService;
    private KafkaConsumerService kafkaConsumerService;




    /**
     * 发送消息（异步）
     * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
     *
     * @param message 消息内容
     */
    @PostMapping("/send")
    @ApiOperation("发送消息（异步）")
    public String sendMessage(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-high", groupId = "group-robotaxi-xpu")
    public void topicProdRobotaxiXpuHigh(String message) {
        try {
            kafkaProducerService.sendMessage(topic, message);
            return "消息发送成功";
            log.info("收到XPU上报高优先级主题消息: {}", message);
            kafkaConsumerService.topicProdRobotaxiXpuHigh(message);
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
            log.error("处理XPU上报高优先级主题消息失败: {}", message, e);
            throw e;
        }
    }

    /**
     * 发送消息（带 Key）
     * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
     *
     * @param message 消息内容
     */
    @PostMapping("/sendWithKey")
    @ApiOperation("发送消息（带 Key）")
    public String sendMessageWithKey(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息 Key", required = true) @RequestParam String key,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
    @KafkaListener(topics = "topic-prod-robotaxi-xpu-normal", groupId = "group-robotaxi-xpu")
    public void topicProdRobotaxiXpuNormal(String message) {
        try {
            kafkaProducerService.sendMessage(topic, key, message);
            return "消息发送成功";
            log.info("收到XPU上报低优先级主题消息: {}", message);
            kafkaConsumerService.topicProdRobotaxiXpuNormal(message);
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送消息（带回调）
     */
    @PostMapping("/sendWithCallback")
    @ApiOperation("发送消息（带回调）")
    public String sendMessageWithCallback(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessageWithCallback(topic, message);
            return "消息发送成功（带回调）";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送消息（同步）
     */
    @PostMapping("/sendSync")
    @ApiOperation("发送消息（同步）")
    public String sendMessageSync(
            @ApiParam(value = "主题", required = true) @RequestParam String topic,
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessageSync(topic, message);
            return "消息发送成功（同步）";
        } catch (Exception e) {
            log.error("发送消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送自动驾驶测试消息
     */
    @PostMapping("/sendAutoPilot")
    @ApiOperation("发送自动驾驶测试消息")
    public String sendAutoPilotMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("automatic-pilot-topic", message);
            return "自动驾驶消息发送成功";
        } catch (Exception e) {
            log.error("发送自动驾驶消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送车辆状态测试消息
     */
    @PostMapping("/sendVehicleStatus")
    @ApiOperation("发送车辆状态测试消息")
    public String sendVehicleStatusMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("vehicle-status-topic", message);
            return "车辆状态消息发送成功";
        } catch (Exception e) {
            log.error("发送车辆状态消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送导航测试消息
     */
    @PostMapping("/sendNavigation")
    @ApiOperation("发送导航测试消息")
    public String sendNavigationMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("navigation-topic", message);
            return "导航消息发送成功";
        } catch (Exception e) {
            log.error("发送导航消息失败", e);
            return "消息发送失败: " + e.getMessage();
        }
    }

    /**
     * 发送订单测试消息
     */
    @PostMapping("/sendOrder")
    @ApiOperation("发送订单测试消息")
    public String sendOrderMessage(
            @ApiParam(value = "消息内容", required = true) @RequestParam String message) {
        try {
            kafkaProducerService.sendMessage("order-topic", message);
            return "订单消息发送成功";
        } catch (Exception e) {
            log.error("发送订单消息失败", e);
            return "消息发送失败: " + e.getMessage();
            log.error("处理XPU上报低优先级主题消息失败: {}", message, e);
            throw e;
        }
    }
}

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

@@ -1,8 +1,48 @@
package com.xiaopeng.automaticPilot.service;

import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson2.JSON;
import com.xiaopeng.api.dto.map.OrderRouteAndFareDto;
import com.xiaopeng.api.dto.operation.LonLatDto;
import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO;
import com.xiaopeng.api.dto.pushMessage.SendMessageDTO;
import com.xiaopeng.api.feignClient.map.XpMapClient;
import com.xiaopeng.api.feignClient.order.XpOrderClient;
import com.xiaopeng.api.feignClient.pushMessage.XpPushMessageClient;
import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient;
import com.xiaopeng.api.model.order.XpOrder;
import com.xiaopeng.api.model.vehicleConvoy.XpVehicle;
import com.xiaopeng.api.vo.automaticPilot.VehiclePosition;
import com.xiaopeng.api.vo.automaticPilot.proto.down.GeoPoint;
import com.xiaopeng.api.vo.automaticPilot.proto.down.cloudOrderOperation.*;
import com.xiaopeng.api.vo.automaticPilot.proto.up.MessageType;
import com.xiaopeng.api.vo.automaticPilot.proto.up.MsgContent;
import com.xiaopeng.api.vo.automaticPilot.proto.up.RTCmdType;
import com.xiaopeng.api.vo.automaticPilot.proto.up.RtMsgHeader;
import com.xiaopeng.api.vo.automaticPilot.proto.up.systemStatusXpu.*;
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.NavigationRequest;
import com.xiaopeng.api.vo.automaticPilot.proto.up.xpuNavigationRequest.OrderPhase;
import com.xiaopeng.api.vo.operation.RouteVo;
import com.xiaopeng.automaticPilot.util.XPUUtil;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.common.core.utils.uuid.UUID;
import com.xiaopeng.common.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.geo.GeoJsonPoint;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * Kafka 消费者服务
@@ -11,111 +51,384 @@
@Service
public class KafkaConsumerService {

    @Resource
    private XpMapClient xpMapClient;

    @Resource
    private XpOrderClient xpOrderClient;

    @Resource
    private MongoTemplate mongoTemplate;

    @Resource
    private XpVehicleClient xpVehicleClient;

    @Resource
    private XPUUtil xpuUtil;

    @Autowired
    public RedisTemplate redisTemplate;

    @Resource
    private XpPushMessageClient xpPushMessageClient;


    /**
     * 消费消息 - 自动驾驶主题
     * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级）
     *
     * @param message 消息内容
     */
    @KafkaListener(topics = "automatic-pilot-topic", groupId = "xp-automaticPilot-group")
    public void consumeAutomaticPilotMessage(String message) {
        try {
            log.info("收到自动驾驶主题消息: {}", message);
            // 处理消息逻辑
            processMessage(message);
        } catch (Exception e) {
            log.error("处理消息失败: {}", message, e);
            throw e;
    public void topicProdRobotaxiXpuHigh(String message) {
        MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
        RtMsgHeader header = msgContent.getHeader();
        MessageType type = header.getType();
        //处理请求消息
        if (MessageType.MESSAGE_TYPE_REQUEST == type) {
            //区分业务类型
            RTCmdType cmd = header.getCmd();
            switch (cmd) {
                case SYSTEM_STATUS_XPU:
                    log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
                    break;
                case XPU_NAVIGATION_REQUEST:
                    log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
                    xpuNavigationRequest(msgContent);
                    break;
                case TRIP_STATUS_UPDATE_EVENT:
                    log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PULLOVER_STATUS_EVENT:
                    log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DRIVING_MODE_UPDATE_EVENT:
                    log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case SYSTEM_STATUS_UPDATE_EVENT:
                    log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case TASK_STATE_UPDATE_EVENT:
                    log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_EVENT:
                    log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case POINT_ARRIVAL_STATUS_EVENT:
                    log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PICKUP_CONFIRMATION_EVENT:
                    log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DROPOFF_INFO_EVENT:
                    log.info("用户下车信息(XPU→云端)：{}", message);
                    break;
                case NAVIGATION_DEVIATION_EVENT:
                    log.info("导航偏航(XPU→云端)：{}", message);
                    break;
                case EMERGENCY_STOP_EVENT:
                    log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case NAVIGATION_UPDATE_EVENT:
                    log.info("导航信息变化(XPU→云端)：{}", message);
                    break;
                case TASK_QUEUE_UPDATE_EVENT:
                    log.info("XPU待执行任务队列(XPU→云端)：{}", message);
                    break;
                case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
                    log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_TO_CDU_EVENT:
                    log.info("更新订单状态到CDU (XPU→云端)：{}", message);
                    break;
                default:
                    log.info("消息不在处理范围内，忽略：{}", message);
            }



        }

    }

    /**
     * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级）
     */
    public void topicProdRobotaxiXpuNormal(String message) {
        MsgContent msgContent = JSON.parseObject(message, MsgContent.class);
        RtMsgHeader header = msgContent.getHeader();
        MessageType type = header.getType();
        //处理请求消息
        if (MessageType.MESSAGE_TYPE_REQUEST == type) {
            //区分业务类型
            RTCmdType cmd = header.getCmd();
            switch (cmd) {
                case SYSTEM_STATUS_XPU:
                    log.info("XPU发布系统运行状态 (XPU→云端)：{}", message);
                    systemStatusXpu(msgContent);
                    break;
                case XPU_NAVIGATION_REQUEST:
                    log.info("XPU向云端请求导航 (XPU→云端)：{}", message);
                    xpuNavigationRequest(msgContent);
                    break;
                case TRIP_STATUS_UPDATE_EVENT:
                    log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PULLOVER_STATUS_EVENT:
                    log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DRIVING_MODE_UPDATE_EVENT:
                    log.info("驾驶模式变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case SYSTEM_STATUS_UPDATE_EVENT:
                    log.info("系统状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case TASK_STATE_UPDATE_EVENT:
                    log.info("任务状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_EVENT:
                    log.info("订单状态变化(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case POINT_ARRIVAL_STATUS_EVENT:
                    log.info("到达目标点状态(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case PICKUP_CONFIRMATION_EVENT:
                    log.info("用户上车确认(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case DROPOFF_INFO_EVENT:
                    log.info("用户下车信息(XPU→云端)：{}", message);
                    break;
                case NAVIGATION_DEVIATION_EVENT:
                    log.info("导航偏航(XPU→云端)：{}", message);
                    break;
                case EMERGENCY_STOP_EVENT:
                    log.info("急停(XPU→云端)(XPU→座舱)：{}", message);
                    break;
                case NAVIGATION_UPDATE_EVENT:
                    log.info("导航信息变化(XPU→云端)：{}", message);
                    break;
                case TASK_QUEUE_UPDATE_EVENT:
                    log.info("XPU待执行任务队列(XPU→云端)：{}", message);
                    break;
                case SYSTEM_STATE_ANAPSHOT_SYNC_EVENT:
                    log.info("每次MQTT连接之后触发 (XPU→云端)：{}", message);
                    break;
                case ORDER_STATE_UPDATE_TO_CDU_EVENT:
                    log.info("更新订单状态到CDU (XPU→云端)：{}", message);
                    break;
                default:
                    log.info("消息不在处理范围内，忽略：{}", message);
            }


        }

    }


    /**
     * XPU发布系统运行状态---车辆实时位置
     * @param msgContent
     */
    public void systemStatusXpu(MsgContent msgContent){
        SystemRuntimeStatus systemRuntimeStatus = msgContent.getContent_detail().getObject("systemRuntimeStatus", SystemRuntimeStatus.class);
        //获取实时位姿
        RealtimePose pose = systemRuntimeStatus.getPose();
        Wgs naviPose = pose.getNavi_pose();
        Double latitude = naviPose.getLatitude();
        Double longitude = naviPose.getLongitude();
        Double course = naviPose.getCourse();
        Float alt = naviPose.getAlt();
        //获取车辆静态信息
        VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicle_status().getStatic_info();
        String vin = staticInfo.getVin();
        redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new GeoJsonPoint(longitude, latitude), vin);

        //保存车辆实时信息到mongodb
        VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class);
        if(null == vehiclePosition){
            vehiclePosition = new VehiclePosition();
        }
        vehiclePosition.setVin(vin);
        vehiclePosition.setPoint(new GeoJsonPoint(longitude, latitude));
        vehiclePosition.setCourse(course);
        vehiclePosition.setAlt(alt);
        VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicle_status();
        //车辆动态信息
        VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamic_status();
        dynamicStatus.setSpeed(dynamicStatus.getSpeed());
        vehiclePosition.setCurrentElectricity(dynamicStatus.getBattery_level());
        //订单信息
        OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunning_order_status_info();
        Long orderId = null;
        String order_id = runningOrderStatusInfo.getOrder_id();
        vehiclePosition.setHasOrder(0);
        if(StringUtils.isNotEmpty(order_id)){
            if(order_id.contains("XU")){
                vehiclePosition.setHasOrder(1);
                orderId = Long.parseLong(order_id.substring(2));
            }else{
                orderId = Long.parseLong(order_id);
            }
        }
        vehiclePosition.setOrderId(orderId);
        mongoTemplate.save(vehiclePosition);
        //调用推送服务接口
        if(null != orderId){
            RtTripInfo rtTripInfo = systemRuntimeStatus.getRt_trip_info();
            PushOrderRouteDTO dto = new PushOrderRouteDTO();
            dto.setOrderId(orderId);
            dto.setRouteType(order_id.contains("XU") ? 2 : 1);
            dto.setLongitude(longitude);
            dto.setLatitude(latitude);
            dto.setDuration(rtTripInfo.getRemaining_time());
            dto.setDistance(rtTripInfo.getRemaining_distance().intValue());
            xpPushMessageClient.pushOrderRoute(dto);
        }
    }





    /**
     * 消费消息 - 车辆状态主题
     *
     * @param message 消息内容
     * XPU向云端请求导航
     * @param msgContent
     */
    @KafkaListener(topics = "vehicle-status-topic", groupId = "xp-automaticPilot-group")
    public void consumeVehicleStatusMessage(String message) {
        try {
            log.info("收到车辆状态主题消息: {}", message);
            // 处理车辆状态消息逻辑
            processVehicleStatus(message);
        } catch (Exception e) {
            log.error("处理车辆状态消息失败: {}", message, e);
            throw e;
    public void xpuNavigationRequest(MsgContent msgContent){
        NavigationRequest naviRequest = msgContent.getContent_detail().getObject("naviRequest", NavigationRequest.class);
        String orderId = naviRequest.getOrder_id();
        String taskId = naviRequest.getTask_id();
        OrderPhase orderPhase = naviRequest.getOrder_phase();
        log.info("订单ID：{}", orderId);
        log.info("任务ID：{}", taskId);
        switch (orderPhase){
            case IDLE:
                log.info("订单的任务未开始执行");
                break;
            case ROUTE_TO_PICKUP:
                log.info("驶往上车点的行程");
                buildNaviRequest(orderId, taskId, orderPhase);
                break;
            case ROUTE_TO_DESTINATION:
                log.info("驶往下车点的行程");
                break;
            case ROUTE_TO_CRUISE:
                log.info("巡游阶段路线");
                break;
        }
    }


    /**
     * 消费消息 - 导航主题
     *
     * @param message 消息内容
     * 构建导航请求
     * @param orderId
     * @param taskId
     * @param orderPhase
     */
    @KafkaListener(topics = "navigation-topic", groupId = "xp-automaticPilot-group")
    public void consumeNavigationMessage(String message) {
        try {
            log.info("收到导航主题消息: {}", message);
            // 处理导航消息逻辑
            processNavigation(message);
        } catch (Exception e) {
            log.error("处理导航消息失败: {}", message, e);
            throw e;
    public void buildNaviRequest(String orderId, String taskId, OrderPhase orderPhase){
        //订单类型  1：乘客订单，2=巡游订单
        Integer orderType = 1;
        LonLatDto destination = new LonLatDto();
        LonLatDto origin = new LonLatDto();
        if(orderId.contains("XU")){
            orderType = 2;
            // todo 下发巡游单的导航数据
            //巡游单
            if(orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)){
                origin.setLongitude(116.40528);
                origin.setLatitude(39.90498);
                destination.setLongitude(116.40528);
                destination.setLatitude(39.90498);
            }
        }else{
            //正常单
            R<XpOrder> xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId));
            if(200 != xpOrderR.getCode()){
                throw new RuntimeException("订单不存在");
            }
            XpOrder xpOrder = xpOrderR.getData();
            //驶往上车点的行程
            if(orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)){
                XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData();
                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(xpOrder.getStartLon());
                destination.setLatitude(xpOrder.getStartLat());
            }
            //驶往下车点的行程
            if(orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)){
                //查询起点到终点的线路
                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);
            R<RouteVo> routeVoR = xpMapClient.getOrderDistanceAndDuration(dto);
            if(200 != routeVoR.getCode()){
                throw new RuntimeException("获取订单线路和运费失败");
            }
            RouteVo routeVo = routeVoR.getData();
            //构建导航下发请求对象
            cloudOrderOperation(orderType, orderId, taskId, routeVo);
        }
    }


    /**
     * 消费消息 - 订单主题
     *
     * @param message 消息内容
     * 云端订单操作----下发导航数据
     * @param routeVo
     */
    @KafkaListener(topics = "order-topic", groupId = "xp-automaticPilot-group")
    public void consumeOrderMessage(String message) {
        try {
            log.info("收到订单主题消息: {}", message);
            // 处理订单消息逻辑
            processOrder(message);
        } catch (Exception e) {
            log.error("处理订单消息失败: {}", message, e);
            throw e;
    public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo){
        //构建消息体
        OrderOperation orderOperation = new OrderOperation();
        orderOperation.setOrder_id(orderId);
        orderOperation.setTask_id(taskId);
        orderOperation.setOperation_type(OrderOperationType.CLOUD_UPDATE_WAYPOINTS);

        //构建航点
        List<Waypoint> waypoints = new ArrayList<>();
        List<LonLatDto> polyline = routeVo.getPolyline();
        for (int i = 0; i < polyline.size(); i++) {
            LonLatDto lonLatDto = polyline.get(i);
            Waypoint waypoint = new Waypoint();
            waypoint.setWaypoint_id(UUID.randomUUID().toString());
            if(0 == i){
                waypoint.setPoint_type(PointType.START_POINT);
                //订单类型不同，点位的动作类型不同
                waypoint.setAction_type(1 == orderType ? ActionType.PICK_UP : ActionType.ACTION_UNSPECIFIED);
            }else if(polyline.size() - 1 == i){
                waypoint.setPoint_type(PointType.END_POINT);
                waypoint.setAction_type(1 == orderType ? ActionType.DROP_OFF : ActionType.ACTION_UNSPECIFIED);
            }else{
                waypoint.setPoint_type(PointType.WAY_POINT);
                //订单类型不同，点位的动作类型不同
                waypoint.setAction_type(ActionType.ACTION_UNSPECIFIED);
            }
            //构建经纬度
            GeoPoint geoPoint = new GeoPoint();
            geoPoint.setLongitude(lonLatDto.getLongitude());
            geoPoint.setLatitude(lonLatDto.getLatitude());
            waypoint.setPoint(geoPoint);
            waypoint.setSequence(i + 1);
            waypoints.add(waypoint);
        }
        orderOperation.setWaypoints(waypoints);
        //开发发送请求到XPU
        xpuUtil.cloudOrderOperation(orderOperation);
    }

    /**
     * 处理消息逻辑（示例）
     *
     * @param message 消息内容
     */
    private void processMessage(String message) {
        // TODO: 实现具体的消息处理逻辑
        log.info("处理自动驾驶消息: {}", message);

    public void c(MsgContent msgContent){



    }

    /**
     * 处理车辆状态消息（示例）
     *
     * @param message 消息内容
     */
    private void processVehicleStatus(String message) {
        // TODO: 实现具体的车辆状态处理逻辑
        log.info("处理车辆状态消息: {}", message);
    }

    /**
     * 处理导航消息（示例）
     *
     * @param message 消息内容
     */
    private void processNavigation(String message) {
        // TODO: 实现具体的导航处理逻辑
        log.info("处理导航消息: {}", message);
    }

    /**
     * 处理订单消息（示例）
     *
     * @param message 消息内容
     */
    private void processOrder(String message) {
        // TODO: 实现具体的订单处理逻辑
        log.info("处理订单消息: {}", message);
    }
}

 xp-robotaxi-automaticPilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaProducerService.java

File was deleted

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

@@ -1,5 +1,7 @@
package com.xiaopeng.automaticPilot.util;

import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.util.IdUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
@@ -81,8 +83,9 @@
        header.setMsg_id(UUID.randomUUID().toString());
        header.setTimestamp(System.currentTimeMillis());
        header.setCmd(RTCmdType.CLOUD_ORDER_OPERATION);
        header.setSequence(1L);
        header.setType(MessageType.MESSAGE_TYPE_RESPONSE);
        Snowflake snowflake = IdUtil.getSnowflake(0, 0);
        header.setSequence(snowflake.nextId());
        header.setType(MessageType.MESSAGE_TYPE_REQUEST);
        msgContent.setHeader(header);
        // 消息内容
        JSONObject jsonObject = new JSONObject();
@@ -108,4 +111,7 @@
    }





}

 xp-robotaxi-order-boot/pom.xml

@@ -27,7 +27,6 @@
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
        </dependency>

        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-seata</artifactId>

 xp-robotaxi-pushMessage-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java

@@ -32,6 +32,8 @@
    private RedisService redisService;
    @Autowired
    private XpPushMessageClient xpPushMessageClient;
    @Autowired
    private XpOrderClient xpOrderClient;

    /**
     * 通过定位推送订单轨迹
@@ -40,9 +42,13 @@
    @PostMapping("/inner/v1/order/pushOrderRoute")
    public R<String> pushOrderRoute(@Validated @RequestBody PushOrderRouteDTO dto) {

        // 获取路线规划
        RouteVo routeVo = redisService.getCacheObject(CacheConstants.ORDER_ROUTE_CAPTURE + "_" + dto.getOrderId() + "_" +dto.getRouteType());

        if(Objects.nonNull(routeVo)){

            // 查询订单信息
            XpOrder xpOrder = xpOrderClient.queryOrderById(dto.getOrderId()).getData();

            // 计算实时点位到线段的最优截取
            List<LonLatDto> lonLatDtos = MapMatchingUtil.clipBySegmentDistance(routeVo.getPolyline(), new LonLatDto(dto.getLongitude(), dto.getLatitude()));
@@ -67,9 +73,15 @@
            route.put("polyline",JSONObject.toJSONString(polyline));

            // 封装消息内容
            Integer routeType = dto.getRouteType();
            if(routeType == 2){
                routeType = 3;
            }else {
                routeType = xpOrder.getStatus() == 4 ? 1 : 2;
            }
            Map<String, Object> content = new HashMap<>();
            content.put("orderId",dto.getOrderId());
            content.put("routeType",dto.getRouteType());
            content.put("routeType",routeType);
            content.put("route",JSONObject.toJSONString(route));

            // 封装轨迹信息
@@ -80,7 +92,7 @@

            // 进行消息推送
            SendMessageDTO sendMessageDTO = new SendMessageDTO();
            sendMessageDTO.setAppUserId(String.valueOf(dto.getAppUserId()));
            sendMessageDTO.setAppUserId(String.valueOf(xpOrder.getAppUserId()));
            sendMessageDTO.setMessage(JSONObject.toJSONString(params));
            xpPushMessageClient.sendRoute(sendMessageDTO);
        }

 xp-robotaxi-vehicleConvoy-boot/src/main/java/com/xiaopeng/vehicleConvoy/util/VehicleControlUtil.java

@@ -46,7 +46,7 @@
     * 测试车辆：L1NNSGHA5PB000204
     * @return
     */
    public VehicleTemperatureDto queryVehicleTemperature(Long uid, String vin) {
    public VehicleTemperatureDto queryVehicleTemperature(String uid, String vin) {
        String nonce = System.currentTimeMillis() + "";
        VehicleTemperatureVo vo = new VehicleTemperatureVo();
        vo.setVin(vin);