.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/factory/vehicleConvoy/XpVehicleClientFallbackFactory.java
@@ -3,6 +3,7 @@ import com.xiaopeng.api.dto.operation.VehiclePositionByVinDto; import com.xiaopeng.api.feignClient.vehicleConvoy.XpVehicleClient; import com.xiaopeng.api.model.vehicleConvoy.XpVehicle; import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.operation.VehiclePositionVo; import com.xiaopeng.common.core.domain.R; import lombok.extern.slf4j.Slf4j; @@ -60,6 +61,12 @@ log.error("查询车辆信息失败，原因: {}", cause.getMessage()); return R.fail(); } @Override public R<XpVehicle> findByVehicleVin(String vehicleId) { log.error("查询车辆信息失败，原因: {}", cause.getMessage()); return R.fail(); } }; } } 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/feignClient/vehicleConvoy/XpVehicleClient.java
@@ -3,6 +3,7 @@ import com.xiaopeng.api.dto.operation.VehiclePositionByVinDto; import com.xiaopeng.api.factory.vehicleConvoy.XpVehicleClientFallbackFactory; import com.xiaopeng.api.model.vehicleConvoy.XpVehicle; import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.operation.VehiclePositionVo; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.dto.ServiceNameConstantsEnum; @@ -71,4 +72,13 @@ */ @PostMapping("/inner/vehicle/queryVehiclePositionByVin") R<List<String>> queryVehiclePositionByVin(@RequestBody VehiclePositionByVinDto dto); /** * 根据车辆vin查询车辆信息 * * @param vehicleVin * @return */ @GetMapping("/inner/vehicle/findByVehicleVin") R<XpVehicle> findByVehicleVin(String vehicleVin); } 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-operation-boot/src/main/java/com/xiaopeng/operation/utils/TaskOrderUtil.java
@@ -13,6 +13,7 @@ import com.xiaopeng.api.model.operation.XpElectronicFence; import com.xiaopeng.api.model.operation.XpTradingStrategy; 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.operation.RouteVo; import com.xiaopeng.api.vo.vehicleConvoy.AddTaxiPassengerVo; @@ -239,14 +240,14 @@ List<VehiclePosition> positionList = vehiclePositionService.findAvailableVehiclesNearby(Double.parseDouble(split[0]), Double.parseDouble(split[1]), xpTradingStrategy.getOnePushDistance(), xpTradingStrategy.getSimultaneouslyOrderCount()); // 获取车辆的区域信息 if(positionList.isEmpty()){ if (positionList.isEmpty()) { continue; } List<String> vinList = positionList.stream().map(VehiclePosition::getVin).collect(Collectors.toList()); // 获取车辆的信息 筛选出可用车辆 R<List<String>> vehiclePositionVos = xpVehicleClient.queryVehiclePositionByVin(new VehiclePositionByVinDto(xpElectronicFence.getId(), vinList, xpTradingStrategy.getSimultaneouslyOrderCount())); if(vehiclePositionVos.getCode()!=200){ R<List<String>> vehiclePositionVos = xpVehicleClient.queryVehiclePositionByVin(new VehiclePositionByVinDto(xpElectronicFence.getId(), vinList, xpTradingStrategy.getSimultaneouslyOrderCount())); if (vehiclePositionVos.getCode() != 200) { continue; } @@ -254,7 +255,7 @@ List<String> data = vehiclePositionVos.getData(); // 找出符合条件的 availableVehiclesNearby = availableVehiclesNearby.stream().filter(e->data.contains(e.getVin())).collect(Collectors.toList()); availableVehiclesNearby = availableVehiclesNearby.stream().filter(e -> data.contains(e.getVin())).collect(Collectors.toList()); // List<VehiclePositionVo> availableVehiclesNearby =vehiclePositionVos.getData(); @@ -384,7 +385,7 @@ if (max1.equals(weightElectricity)) { // 计算车的分数 分数最大是车的size availableVehiclesNearby= availableVehiclesNearby.stream().sorted(Comparator.comparing(VehiclePositionVo::getCurrentElectricity).reversed()).collect(Collectors.toList()); availableVehiclesNearby = availableVehiclesNearby.stream().sorted(Comparator.comparing(VehiclePositionVo::getCurrentElectricity).reversed()).collect(Collectors.toList()); int size = tendencyElectricity; // 做比对的分数 int temp = availableVehiclesNearby.get(0).getCurrentElectricity(); @@ -399,9 +400,9 @@ // 算出总分 再来排序计算 for (VehiclePositionVo vehiclePositionVo : availableVehiclesNearby) { vehiclePositionVo.setOrderValueTotal(vehiclePositionVo.getCurrentElectricityScore() + vehiclePositionVo.getDistanceScore() + vehiclePositionVo.getTimeScore() + vehiclePositionVo.getDistanceDifference()+ vehiclePositionVo.getTimeDifference()); vehiclePositionVo.setOrderValueTotal(vehiclePositionVo.getCurrentElectricityScore() + vehiclePositionVo.getDistanceScore() + vehiclePositionVo.getTimeScore() + vehiclePositionVo.getDistanceDifference() + vehiclePositionVo.getTimeDifference()); } availableVehiclesNearby= availableVehiclesNearby.stream().sorted(Comparator.comparing(VehiclePositionVo::getOrderValueTotal).reversed()).collect(Collectors.toList()); availableVehiclesNearby = availableVehiclesNearby.stream().sorted(Comparator.comparing(VehiclePositionVo::getOrderValueTotal).reversed()).collect(Collectors.toList()); int size = tendencyOrderValue; @@ -517,7 +518,7 @@ Integer secondMax = values.get(1); // 判断第二大的是哪个 if(Objects.equals(secondMax, tendencyDistance)){ if (Objects.equals(secondMax, tendencyDistance)) { for (VehiclePositionVo vehiclePosition : availableVehiclesNearby) { RouteAndFareDto routeAndFareDto = new RouteAndFareDto(); LonLatDto lonLatDto = new LonLatDto(); @@ -551,7 +552,7 @@ } if(Objects.equals(secondMax, tendencyTime)){ if (Objects.equals(secondMax, tendencyTime)) { for (VehiclePositionVo vehiclePosition : availableVehiclesNearby) { RouteAndFareDto routeAndFareDto = new RouteAndFareDto(); LonLatDto lonLatDto = new LonLatDto(); @@ -586,7 +587,7 @@ // 订单价值 if(Objects.equals(secondMax, tendencyOrderValue)){ if (Objects.equals(secondMax, tendencyOrderValue)) { } @@ -601,31 +602,33 @@ } VehiclePositionVo vehiclePositionVo = availableVehiclesNearby.stream().sorted(Comparator.comparing(VehiclePositionVo::getTotal).reversed()).collect(Collectors.toList()).get(0); VehiclePositionVo vehiclePositionVo1 = vehiclePositionVo; RLock lock = redissonClient.getLock("FIND_CAR_" + vehiclePositionVo1.getVin()); // // VehiclePositionVo vehiclePositionVo1 = vehiclePositionVo; RLock lock = redissonClient.getLock("FIND_CAR_" + vehiclePositionVo.getVin()); lock.lock(10, TimeUnit.SECONDS); try { VehiclePosition vehiclePosition = vehiclePositionService.findByVehicleId(vehiclePositionVo.getVehicle_id()); // VehiclePosition vehiclePosition = vehiclePositionService.findByVehicleId(vehiclePositionVo.getVehicle_id()); R<XpVehicle> vehiclePositionR = xpVehicleClient.findByVehicleVin(vehiclePositionVo.getVin()); if (vehiclePositionR.getCode() != 200) { continue; } XpVehicle vehiclePosition = vehiclePositionR.getData(); if (vehiclePosition.getStatus() != 1 || vehiclePosition.getState() != 0) { lock.unlock(); continue; } if (vehiclePosition.getHasOrder() == 1) { // 关闭巡游单子 TODO } // 关闭巡游单子 TODO AddTaxiPassengerVo addTaxiPassengerVo = new AddTaxiPassengerVo(); addTaxiPassengerVo.setVin(vehiclePosition.getVin()); addTaxiPassengerVo.setVin(vehiclePosition.getVehicleVin()); addTaxiPassengerVo.setUid(uid); addTaxiPassengerVo.setOrderNo(orderNo); // TODO appid addTaxiPassengerVo.setAppId(""); BaseVehicleUserDto baseVehicleUserDto = userVehicleClient.addTaxiPassenger(addTaxiPassengerVo); if (baseVehicleUserDto.getCode() == 200) { // TODO 下单 // 绑定成功 弹出这个队列第一个订单 redisService.popFirstElement(quoteName); @@ -640,13 +643,17 @@ } else { // 直接派单 定义一个锁来锁定车辆 VehiclePosition vehiclePosition = positionList.get(0); Long vehicleId = vehiclePosition.getVehicle_id(); String vehicleVin = vehiclePosition.getVin(); RLock lock = redissonClient.getLock("FIND_CAR_" + vehiclePosition.getVin()); lock.lock(10, TimeUnit.SECONDS); try { // 查询车辆状态 VehiclePosition vehicleId1 = vehiclePositionService.findByVehicleId(vehicleId); R<XpVehicle> vehicleR = xpVehicleClient.findByVehicleVin(vehicleVin); if(vehicleR.getCode()!=200){ continue; } XpVehicle vehicleId1 = vehicleR.getData(); if (vehicleId1.getStatus() != 1 || vehicleId1.getState() != 0) { lock.unlock(); continue; 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-order-boot/src/main/java/com/xiaopeng/order/controller/XpOrderController.java
@@ -335,7 +335,7 @@ * 手动结束订单 */ @PutMapping("/api/v1/order/endTrip/{order_id}") @PutMapping("/api/v1/mini-program/order/endTrip/{order_id}") @ApiOperation(value = "结束行程", tags = {"小程序-结束行程"}) @XpClientCheck(checkToken = false, checkOtpToken = true) public R endTrip(@PathVariable("orderId") Long orderId,HttpServletRequest request) { @@ -348,7 +348,7 @@ * 获取行程费用明细 */ @PutMapping("/api/v1/order/fare-breakdown/{order_id}") @PutMapping("/api/v1/mini-program/order/fare-breakdown/{order_id}") @ApiOperation(value = "获取行程费用明细", tags = {"小程序-获取行程费用明细"}) @XpClientCheck(checkToken = false, checkOtpToken = true) public R<FareBreakdownVo> fareBreakdown(@PathVariable("orderId") Long orderId, HttpServletRequest request) { @@ -359,7 +359,7 @@ /** * 订单评价 */ @PostMapping("/api/v1/order/payment") @PostMapping("/api/v1/mini-program/order/payment") @ApiOperation(value = "发起支付", tags = {"小程序-发起支付"}) @XpClientCheck(checkToken = false, checkOtpToken = true) public R payment(@Validated @RequestBody XpOrderPayDto dto, HttpServletRequest request) { 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/controller/XpVehicleController.java
@@ -15,6 +15,7 @@ import com.xiaopeng.api.feignClient.order.XpOrderClient; import com.xiaopeng.api.model.vehicleConvoy.XpVehicle; import com.xiaopeng.api.model.vehicleConvoy.XpVehicleAnnualRecord; import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.vehicleConvoy.VehiclesCommandCallbackVo; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.exception.ServiceException; @@ -34,6 +35,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.poi.ss.usermodel.Workbook; import org.springframework.context.annotation.Lazy; import org.springframework.data.mongodb.core.geo.GeoJsonPoint; import org.springframework.util.CollectionUtils; import org.springframework.web.bind.annotation.*; import org.springframework.web.multipart.MultipartFile; @@ -392,6 +394,18 @@ } } /** * 查询车辆信息 * @param vehicleVin * @return */ @GetMapping("/inner/vehicle/findByVehicleVin") public R<XpVehicle> findByVehicleVin(String vehicleVin){ XpVehicle xpVehicle = xpVehicleService.getOne(new LambdaQueryWrapper<XpVehicle>().eq(XpVehicle::getVehicleVin, vehicleVin)); return R.ok(xpVehicle); } /** * 查询可改派的车辆列表 * 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);
