package com.xiaopeng.automaticPilot.service; import cn.hutool.core.lang.Snowflake; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.util.JsonFormat; import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto; import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO; import com.xiaopeng.api.dto.map.OrderRouteAndFareDto; 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.PopOrderDto; import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO; import com.xiaopeng.api.dto.pushMessage.SendMessageDTO; 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; import com.xiaopeng.api.feignClient.operation.ReassignOrderClient; import com.xiaopeng.api.feignClient.operation.XpCruisePlanClient; 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.order.XpCruiseOrder; import com.xiaopeng.api.model.order.XpOrder; import com.xiaopeng.api.model.order.XpTestOrder; import com.xiaopeng.api.model.vehicleConvoy.XpVehicle; import com.xiaopeng.api.vo.OrderType; import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.automaticPilot.protobuf.*; 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.util.XPUUtil; import com.xiaopeng.common.core.constant.CacheConstants; import com.xiaopeng.common.core.constant.Constants; import com.xiaopeng.common.core.domain.R; 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.StringUtils; import com.xiaopeng.common.core.utils.uuid.UUID; import com.xiaopeng.common.redis.service.RedisService; import lombok.extern.slf4j.Slf4j; import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.geo.Point; 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.mongodb.core.query.Update; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.*; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * Kafka 消费者服务 */ @Slf4j @Service public class KafkaConsumerService { private static final String REASSIGN_CANCEL_CONTEXT_PREFIX = "REASSIGN_CANCEL_CONTEXT:"; @Resource private XpMapClient xpMapClient; @Resource private XpOrderClient xpOrderClient; @Resource private MongoTemplate mongoTemplate; @Resource private XpVehicleClient xpVehicleClient; @Resource private XPUUtil xpuUtil; @Resource public RedisTemplate redisTemplate; @Resource private XpPushMessageClient xpPushMessageClient; @Resource private ReassignOrderClient reassignOrderClient; @Resource private UserVehicleClient userVehicleClient; @Resource private XpCruiseOrderClient xpCruiseOrderClient; @Resource private RedisService redisService; @Resource private XpCruisePlanClient xpCruisePlanClient; @Resource(name = "hutoolSnowflake") private Snowflake snowflake; @Value("${robotaxi.appId:xp_robotaxi_api}") private String appid; @Value("${automatic.content.waittime:20000}") private Long waittime; @Resource private XpTaskClient xpTaskClient; @Resource private VehicleBuriedPointClient vehicleBuriedPointClient; @Value("${xpu.res.ping.enable: true}") private boolean xpuResPingEnable; @Resource private XpTestOrderClient testOrderClient; @Resource private RedissonClient redissonClient; private static final JsonFormat.Printer jsonPrinter = JsonFormat.printer() .printingEnumsAsInts() .preservingProtoFieldNames(); /** * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级） * */ public void topicProdRobotaxiXpuHigh(String message) { MsgContent msgContent = parseMsgContent(message); if (msgContent != null) { handleXPUMessage(msgContent, message); } } /** * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级） */ public void topicProdRobotaxiXpuNormal(String message) { MsgContent msgContent = parseMsgContent(message); if (msgContent != null) { handleXPUMessage(msgContent, message); } } private MsgContent parseMsgContent(String message) { try { // 首先解析外层JSON，提取payload字段 JSONObject outerJson = JSON.parseObject(message); String payload = outerJson.getString("payload"); if (StringUtils.isEmpty(payload)) { log.error("Payload is empty in message: {}", message); return null; } // 使用payload构建MsgContent MsgContent.Builder builder = MsgContent.newBuilder(); JsonFormat.parser().ignoringUnknownFields().merge(payload, builder); return builder.build(); } catch (InvalidProtocolBufferException e) { log.error("Failed to parse message: {}", message, e); return null; } catch (Exception e) { log.error("Failed to extract payload from message: {}", message, e); return null; } } /** * 处理XPU消息 * @param msgContent * @param message */ public void handleXPUMessage(MsgContent msgContent, String message){ MsgHeader header = msgContent.getHeader(); MessageType type = header.getType(); //处理请求消息 if (MessageType.MESSAGE_TYPE_REQUEST == type) { //区分业务类型 int cmdValue = header.getCmd(); RTCmdType cmd = RTCmdType.getByValue(cmdValue); if (cmd == null) { log.warn("Unknown cmd value: {}", cmdValue); return; } switch (cmd) { case SYSTEM_STATUS_XPU: log.info("XPU发布系统运行状态 (XPU→云端)：{}", message); systemStatusXpu(cmd, msgContent); systemStatusXpuAndReassign(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); orderStateUpdateEvent(cmd, msgContent); 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); navigationDeviationEvent(msgContent); 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; case ORDER_NEAR_DESTINATION_EVENT: log.info("车辆即将到达订单终点 (XPU→云端)：{}", message); // if(orderId.contains("XU")){ // buildPushOrderStatus(orderId, OrderStatusEnum.COMPLETED, sequence); // } break; default: log.info("消息不在处理范围内，忽略：{}", message); } } //消息应答 if(MessageType.MESSAGE_TYPE_RESPONSE == type) { //区分业务类型 int cmdValue = header.getCmd(); RTCmdType cmd = RTCmdType.getByValue(cmdValue); if (cmd == null) { log.warn("Unknown cmd value: {}", cmdValue); return; } //云端下发订单的响应报文 if (cmd == RTCmdType.CLOUD_SEND_RT_TASK) { RtTaskListResponse response = msgContent.getRtTaskListResponse(); try { log.info("云端下发订单操作的响应消息 (XPU→云端)：{}", jsonPrinter.print(response)); RtTaskResult rtTaskResult = response.getResults(0); if (rtTaskResult.getStatus() != TaskResultStatus.ACCEPTED) { log.error("订单操作失败，orderId {} taskId {} 错误码 {}", rtTaskResult.getOrderId(), rtTaskResult.getTaskId(), rtTaskResult.getStatus()); return; } Object sequence = redisService.getCacheObject("CREATE_ORDER:" + response.getResults(0).getOrderId()); if(sequence!=null){ JSONObject jsonObject = JSON.parseObject(sequence.toString()); String orderId = jsonObject.getString("orderId"); String vin = jsonObject.getString("vin"); String orderType = jsonObject.getString("orderType"); PopOrderDto dto = new PopOrderDto(); if(orderId.contains("XU")){ dto.setOrderId(Long.valueOf(orderId.substring(2))); }else if(orderId.contains("TEST")){ dto.setOrderId(Long.valueOf(orderId.substring(4))); }else{ dto.setOrderId(Long.valueOf(orderId)); } dto.setXpVehicleVin(vin); dto.setOrderType(orderType); xpOrderClient.popOrder(dto); log.info("订单状态更新：{}", response.getResults(0).getOrderId()); } } catch (InvalidProtocolBufferException e) { log.error("Failed to print RtTaskListResponse: {}", response, e); } } else { log.info("响应消息不在处理范围内，忽略：{}", message); } } } /** * XPU发布系统运行状态---车辆实时位置 * * @param msgContent */ public void systemStatusXpu(RTCmdType cmd, MsgContent msgContent) { //开始处理消息对应的业务逻辑 SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus(); //获取实时位置 RealtimePose pose = systemRuntimeStatus.getPose(); Wgs naviPose = pose.getNaviPose(); double latitude = naviPose.getLatitude(); double longitude = naviPose.getLongitude(); Double course = naviPose.getCourse(); Float alt = naviPose.getAlt(); //获取车辆静态信息 VehicleStaticInfo staticInfo = systemRuntimeStatus.getVehicleStatus().getStaticInfo(); String xpuId = staticInfo.getXpuId(); XpVehicle xpVehicle = xpVehicleClient.findByVehicleXpuId(xpuId).getData(); if(null == xpVehicle){ log.info("XPUID查询车辆无效 {}", xpuId); return; } String vin = xpVehicle.getVehicleVin(); //先响应消息 long sequence = msgContent.getHeader().getSequence(); if (xpuResPingEnable) { responseMessage(cmd, sequence, vin); } VehicleStatus vehicleStatus = systemRuntimeStatus.getVehicleStatus(); OrderStatusInfo runningOrderStatusInfo = systemRuntimeStatus.getRunningOrderStatusInfo(); String order_id = runningOrderStatusInfo.getOrderId(); RLock lock = redissonClient.getLock("PUSH_ORDER_ROUTE_LOCK:" + order_id); try { boolean tryLock = lock.tryLock(10, 10, TimeUnit.SECONDS); if (!tryLock) { log.info("获取锁失败，订单ID：{} 无法完成车辆实时位置的逻辑处理。", order_id); return; } //车辆动态信息 VehicleDynamicStatus dynamicStatus = vehicleStatus.getDynamicStatus(); Long orderId = null; Integer hasOrder = 0; if (StringUtils.isNotEmpty(order_id)) { if (order_id.contains("XU")) { hasOrder = 1; orderId = Long.parseLong(order_id.substring(2)); } else if(order_id.contains("TEST")) { orderId = Long.parseLong(order_id.substring(4)); }else{ orderId = Long.parseLong(order_id); } } //保存车辆实时信息到mongodb if(0 != longitude && 0 != latitude){ //缓存车辆历史定位，用户辅助获取导航抓路用 // List history = redisService.getCacheList("VEHICLE_HISTORY_POINT_LIST:" + xpVehicle.getId()); // if(history.size() >= 10){ // history = new ArrayList<>(history.subList(history.size() - 9, history.size())); // } // HistoryPointDto historyPointDto = new HistoryPointDto(); // historyPointDto.setLocation(longitude + "," + latitude); // historyPointDto.setLocateTime(System.currentTimeMillis() + ""); // historyPointDto.setAccuracy(naviPose.getPosAcc()); // historyPointDto.setDirection(new BigDecimal(course).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue()); // historyPointDto.setSpeed(dynamicStatus.getSpeed()); // historyPointDto.setHeight(naviPose.getAlt()); // history.add(historyPointDto); // redisService.setCacheList("VEHICLE_HISTORY_POINT_LIST:" + xpVehicle.getId(), history); redisTemplate.opsForGeo().add("vehiclePosition:" + vin, new Point(longitude, latitude), vin); 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); vehiclePosition.setSpeed(dynamicStatus.getSpeed()); vehiclePosition.setCurrentElectricity(dynamicStatus.getBmsSocDisp()); vehiclePosition.setMileageRange(dynamicStatus.getDstBatDisp()); vehiclePosition.setOrderId(orderId); vehiclePosition.setTotalOdometer(dynamicStatus.getTotalOdometer()); vehiclePosition.setAddTime(System.currentTimeMillis()); mongoTemplate.save(vehiclePosition); }else{ vehiclePosition.setOrderId(orderId); Update update = new Update() .set("point", new GeoJsonPoint(longitude, latitude)) .set("course", course) .set("alt", alt) .set("speed", dynamicStatus.getSpeed()) .set("currentElectricity", dynamicStatus.getBmsSocDisp()) .set("mileageRange", dynamicStatus.getDstBatDisp()) .set("hasOrder", hasOrder) .set("orderId", orderId) .set("totalOdometer", dynamicStatus.getTotalOdometer()) .set("addTime", System.currentTimeMillis()); mongoTemplate.updateFirst(new Query(Criteria.where("vin").is(vin)), update, VehiclePosition.class); } //调用推送服务接口，给乘客推送剩余轨迹和里程时长 RtTripInfo rtTripInfo = systemRuntimeStatus.getRunningOrderStatusInfo().getRtTripInfo(); if (null != orderId && !order_id.contains("XU")) { PushOrderRouteDTO dto = new PushOrderRouteDTO(); dto.setOrderId(orderId); dto.setRouteType(1); dto.setLongitude(longitude); dto.setLatitude(latitude); if (rtTripInfo.hasCurrentTask() && rtTripInfo.getCurrentTask().hasNaviProgressInfo()) { dto.setDuration(rtTripInfo.getCurrentTask().getNaviProgressInfo().getRouteRemainTime()); dto.setDistance(rtTripInfo.getCurrentTask().getNaviProgressInfo().getRouteRemainDist()); dto.setDirection(course.intValue()); } OrderTrajectoryVo orderTrajectory = new OrderTrajectoryVo(); orderTrajectory.setOrderId(orderId); orderTrajectory.setLocation(longitude + "," + latitude); orderTrajectory.setDirection(dto.getDirection()); orderTrajectory.setSpeed((double) dynamicStatus.getSpeed()); orderTrajectory.setLocatetime(System.currentTimeMillis()); orderTrajectory.setCreateTime(new Date()); //只有乘客订单需要推送剩余导航数据 if(!order_id.contains("TEST")){ xpPushMessageClient.pushOrderRoute(dto);//存储历史轨迹 // 给连环派单的用户推送 Long nextOrderId = redisService.getCacheObject(CacheConstants.nextOrder(order_id)); if (null != nextOrderId){ dto.setOrderId(nextOrderId); xpPushMessageClient.pushOrderRoute(dto); } xpOrderClient.saveVehicleLocation(orderTrajectory); }else{ testOrderClient.saveVehicleLocation(orderTrajectory); } } } }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } /** * XPU发布系统运行状态---车辆实时位置 * * @param msgContent */ public void systemStatusXpuAndReassign(MsgContent msgContent) { //开始处理消息对应的业务逻辑 SystemRuntimeStatus systemRuntimeStatus = msgContent.getSystemRuntimeStatus(); AbnormalReason abnormalReason = systemRuntimeStatus.getSystemStatus().getAbnormalReason(); // 存储车辆挡位 GearLevel gearLevel = systemRuntimeStatus.getVehicleStatus().getDynamicStatus().getGearLevel(); int value = gearLevel.getNumber(); redisService.setCacheObject(CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicleStatus().getStaticInfo().getXpuId(), value); log.info("车辆挡位：{} {}", CacheConstants.GEAR_LEVEL + systemRuntimeStatus.getVehicleStatus().getStaticInfo().getXpuId(), value); // 判断是否为故障状态 XpVehicle vehicle = null; if (abnormalReason == AbnormalReason.FAULT) { log.info("车辆故障，开始处理故障"); String orderId = systemRuntimeStatus.getRunningOrderStatusInfo().getOrderId(); if(orderId.contains("XU")){ //巡游订单，需要关闭订单 xpCruiseOrderClient.cancelCruiseOrderById(Long.parseLong(orderId.substring(2))); XpCruiseOrder cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2))).getData(); vehicle = xpVehicleClient.queryById(cruiseOrder.getVehicleId()).getData(); if (vehicle != null) { // 修改车辆状态 XpVehicle vehicle1 = new XpVehicle(); vehicle1.setOperatingStatus(4); vehicle1.setId(vehicle.getId()); xpVehicleClient.updateVehicleById(vehicle1); } }else if(orderId.contains("TEST")){ //测试订单故障则取消订单 testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), 0); XpTestOrder testOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4))).getData(); vehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData(); if (vehicle != null) { // 修改车辆状态 XpVehicle vehicle1 = new XpVehicle(); vehicle1.setOperatingStatus(4); vehicle1.setId(vehicle.getId()); xpVehicleClient.updateVehicleById(vehicle1); log.info("车辆故障，车辆状态修改0 {}", vehicle.getPlateNumber()); } }else { //正常订单 车辆故障 /* 1.将订单信息车辆的 查询出来 设置故障状态 2.解除用户和车的关系 3.如果是 接驾中的车需要重新派车 4.如果是行驶中的车需要将订单结束 计算金额 */ XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData(); vehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); if (vehicle != null) { // 修改车辆状态 XpVehicle vehicle1 = new XpVehicle(); vehicle1.setState(0);//车辆接单状态 1接单中 0空闲 vehicle1.setOperatingStatus(4);//运营状态 1运营中 2维修中 3已收车 4车辆异常 vehicle1.setId(vehicle.getId()); xpVehicleClient.updateVehicleById(vehicle1); log.info("车辆故障，车辆状态修改0 {}", vehicle.getPlateNumber()); // 解绑用户与车辆关系 CancelTaxiPassengerVo cancelTaxiPassengerVo = new CancelTaxiPassengerVo(); cancelTaxiPassengerVo.setVin(vehicle.getVehicleVin()); cancelTaxiPassengerVo.setUid(xpOrder.getAppUserId().toString()); cancelTaxiPassengerVo.setOrderNo(xpOrder.getOrderNumber()); cancelTaxiPassengerVo.setAppId(appid); userVehicleClient.cancelTaxiPassenger(cancelTaxiPassengerVo); } //status 状态 0=已取消，1=待接单，2=待出发，3=接驾中，4=待客上车，5=乘客已上车，6=服务中，7=待客下车，8=待付款，9=待评价，10=已完成 List notDrive = new ArrayList(){{ //未行驶的情况下订单故障状态点 status=2、3、4、5 add(OrderStatusEnum.READY_DEPART.getCode());add(OrderStatusEnum.PICK_UP.getCode()); add(OrderStatusEnum.PICK_UP_GUESTS.getCode());add(OrderStatusEnum.PASSENGER_BOARDED.getCode()); }}; if(xpOrder.getStatus().equals(OrderStatusEnum.IN_SERVICE.getCode()) ){ //6=服务中 在行驶中 修改订单-已完成 需要处理订单金额 //行驶中车辆异常，订单状态修改- 修改状态=7 再修改 金额 再将状态修改成8 xpOrderClient.carAbnormalUpdateOrderStatus(Long.parseLong(orderId)); }else if(notDrive.contains(xpOrder.getStatus())){//在接驾中 车辆出现故障 重新派单 改派 ReassignOrderDTO reassignOrderDTO = new ReassignOrderDTO(); reassignOrderDTO.setUid(xpOrder.getAppUserId().toString()); reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber()); reassignOrderDTO.setCount(10); // RealtimePose pose = systemRuntimeStatus.getPose(); // Wgs naviPose = pose.getNaviPose(); // Double latitude = naviPose.getLatitude(); // Double longitude = naviPose.getLongitude(); reassignOrderDTO.setLongitude(xpOrder.getStartLon()); reassignOrderDTO.setLatitude(xpOrder.getStartLat()); reassignOrderClient.reassignOrder(reassignOrderDTO); } } //添加车辆异常任务和发送消息 xpTaskClient.saveVehicleAccidentError(vehicle.getId()); // 封装埋点消息实体 VehicleEventDTO vehicleEventDTO = new VehicleEventDTO(); vehicleEventDTO.setVin(vehicle.getVehicleVin()); vehicleEventDTO.setXpuId(""); vehicleEventDTO.setOrderId(""); vehicleEventDTO.setEventKey("RT_web_vehicle_status_changed"); vehicleEventDTO.setEventTime(System.currentTimeMillis()); Map attributes = new HashMap<>(); attributes.put("vin",vehicle.getVehicleVin()); attributes.put("order_id",""); attributes.put("timestamp",vehicleEventDTO.getEventTime()); attributes.put("main_status","OFFLINE"); // TODO 预留位置 目前无状态 attributes.put("sub_status",""); attributes.put("prev_sub_status",""); vehicleEventDTO.setAttributes(attributes); vehicleBuriedPointClient.sendVehicleEvent(vehicleEventDTO); } } /** * XPU向云端请求导航 * * @param msgContent */ public void xpuNavigationRequest(MsgContent msgContent) { //开始处理消息对应的业务逻辑 NavigationRequest naviRequest = msgContent.getNavigationRequest(); String requestId = naviRequest.getRequestId(); String orderId = naviRequest.getOrderId(); String taskId = naviRequest.getTaskId(); OrderPhase orderPhase = naviRequest.getOrderPhase(); log.info("订单ID：{}", orderId); log.info("任务ID：{}", taskId); switch (orderPhase) { case IDLE: log.info("订单的任务未开始执行"); break; case ROUTE_TO_PICKUP: log.info("驶往上车点的行程 {}", orderId); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence(), taskId); break; case ROUTE_TO_DESTINATION: log.info("驶往下车点的行程 {}", orderId); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence(), taskId); break; case ROUTE_TO_CRUISE: log.info("巡游阶段路线 {}", orderId); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence(), taskId); break; } } /** * 车辆偏航后重新规划线路 * * @param msgContent */ public void navigationDeviationEvent(MsgContent msgContent) { //开始处理消息对应的业务逻辑 NavigationDeviationEvent deviationEvent = msgContent.getNaviDeviationEvent(); String requestId = deviationEvent.getNavigationDeviationMessage().getRequestId() + ""; //导航id String routeID = ""; Long vehicleId = null; String orderType = ""; //偏航事件类型 1：请求导航，2：取消导航 NavigationDeviationMessage navigationDeviationMessage = deviationEvent.getNavigationDeviationMessage(); String orderId = navigationDeviationMessage.getOrderId(); String taskId = navigationDeviationMessage.getTaskId(); Wgs wgs = navigationDeviationMessage.getPoint(); Double longitude = wgs.getLongitude(); Double latitude = wgs.getLatitude(); //订单类型 1：乘客订单，2=巡游订单 LonLatDto destination = new LonLatDto(); LonLatDto origin = new LonLatDto(); RouteVo routeVo = null; String key = ""; //添加redis锁，用于在发生偏航后重新规划线路与正常推送剩余导航数据之间的冲突 RLock lock = redissonClient.getLock("PUSH_ORDER_ROUTE_LOCK:" + orderId); try { boolean tryLock = lock.tryLock(10, 10, TimeUnit.SECONDS); if(!tryLock){ log.info("获取锁失败，订单ID：{} 无法完成偏航后的逻辑处理。", orderId); return; } if (orderId.contains("XU")) { orderType = OrderType.CRUISE_ORDER; R cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2))); XpCruiseOrder xpCruiseOrder = cruiseOrder.getData(); //构建导航数据给到车端 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(xpCruiseOrder.getEndLon()); destination.setLatitude(xpCruiseOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(2))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(3); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); vehicleId = xpCruiseOrder.getVehicleId(); } else if(orderId.contains("TEST")){ R cruiseOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4))); XpTestOrder testOrder = cruiseOrder.getData(); //区分测试单和测试巡游单 if(1 == testOrder.getOrderType()){ orderType = OrderType.TEST_ORDER; //驶往上车点的行程 if (3 == testOrder.getStatus()) { //查询司机当前位置到起点的线路 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(testOrder.getStartLon()); destination.setLatitude(testOrder.getStartLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(1); dto.setVehicleId(testOrder.getVehicleId()); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); key = "TEST_ORDER_NAVIGATION_START_DATA:" + testOrder.getVehicleId(); } //驶往下车点的行程 if (6 == testOrder.getStatus()) { origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(testOrder.getEndLon()); destination.setLatitude(testOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(2); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); key = "TEST_ORDER_NAVIGATION_END_DATA:" + testOrder.getVehicleId(); } }else{ orderType = OrderType.TEST_CRUISE_ORDER; //构建导航数据给到车端 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(testOrder.getEndLon()); destination.setLatitude(testOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(3); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); vehicleId = testOrder.getVehicleId(); } } else { orderType = OrderType.USER_ORDER; //正常单 R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); if (200 != xpOrderR.getCode()) { throw new RuntimeException("订单不存在"); } XpOrder xpOrder = xpOrderR.getData(); XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); //驶往上车点的行程 if (3 == xpOrder.getStatus()) { //查询司机当前位置到起点的线路 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(xpOrder.getStartLon()); destination.setLatitude(xpOrder.getStartLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId)); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(1); dto.setVehicleId(xpOrder.getVehicleId()); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); key = "ORDER_NAVIGATION_START_DATA:" + xpOrder.getAppUserId(); } //驶往下车点的行程 if (6 == xpOrder.getStatus()) { origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(xpOrder.getEndLon()); destination.setLatitude(xpOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId)); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(2); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); key = "ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId(); } vehicleId = xpOrder.getVehicleId(); } //直接响应回最新导航数据 if(StringUtils.isNotEmpty(routeID)){ //推送小程序新的导航数据 if(orderType.equals(OrderType.USER_ORDER)){ //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 redisService.setCacheObject(key, JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); PushOrderRouteDTO dto = new PushOrderRouteDTO(); dto.setOrderId(Long.parseLong(orderId)); dto.setRouteType(1); dto.setLongitude(longitude); dto.setLatitude(latitude); if (null != routeVo) { dto.setDuration(null == routeVo.getDuration() ? 0 : routeVo.getDuration().doubleValue()); dto.setDistance(null == routeVo.getDistance() ? 0 : routeVo.getDistance().doubleValue()); dto.setDirection(null == routeVo.getDirection() ? 0 : routeVo.getDirection().intValue()); } dto.setMessageType(PushMessageTypeEnum.ORDER_UPDATE_LINE.getDesc()); xpPushMessageClient.pushOrderRoute(dto); log.info("完成小程序下发导航偏离后的新轨迹：" + orderId); } XpVehicle xpVehicle = xpVehicleClient.queryById(vehicleId).getData(); R stringR = xpMapClient.queryRouted(routeID); if (200 != stringR.getCode()) { throw new RuntimeException("根据导航ID获取车端需要的导航数据失败"); } log.info("开始下发XPU导航偏离后新的导航数据：" + orderId); //构建响应导航数据给车端 cloudPushNavigationInfo(requestId, routeID, stringR.getData(), orderType, orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId(), taskId); log.info("完成下发XPU导航偏离后新的导航数据：" + orderId); } }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } /** * 构建导航请求 * * @param orderId * @param orderPhase */ public void buildNaviRequest(String requestId, String orderId, OrderPhase orderPhase, Long sequence, String taskId) { //订单类型 1：乘客订单，2=巡游订单 LonLatDto destination = new LonLatDto(); LonLatDto origin = new LonLatDto(); //导航id String routeID = ""; Long vehicleId = null; String orderType = ""; if (orderId.contains("XU")) { orderType = OrderType.CRUISE_ORDER; //巡游单 if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) { /* 构建导航数据给到车端 */ R cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2))); XpCruiseOrder xpCruiseOrder = cruiseOrder.getData(); origin.setLongitude(xpCruiseOrder.getStartLon()); origin.setLatitude(xpCruiseOrder.getStartLat()); destination.setLongitude(xpCruiseOrder.getEndLon()); destination.setLatitude(xpCruiseOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(2))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(3); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); vehicleId = xpCruiseOrder.getVehicleId(); } } else if(orderId.contains("TEST")){ R cruiseOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4))); XpTestOrder testOrder = cruiseOrder.getData(); XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData(); //区分测试单和测试巡游单 if(1 == testOrder.getOrderType()){ orderType = OrderType.TEST_ORDER; //驶往上车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) { //直接取缓存中的导航数据推送给车端和乘客 Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + testOrder.getVehicleId()); if(null == cacheObject){ //没有缓存数据，重新规划线路 //查询车辆位置 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(testOrder.getStartLon()); destination.setLatitude(testOrder.getStartLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(2); dto.setVehicleId(testOrder.getVehicleId()); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 redisService.setCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + testOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); }else{ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); } } //驶往下车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) { //直接取缓存中的导航数据推送给车端和乘客 Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + testOrder.getVehicleId()); if(null == cacheObject){ //没有缓存数据，重新规划线路 //查询起点到终点的线路 origin.setLongitude(testOrder.getStartLon()); origin.setLatitude(testOrder.getStartLat()); destination.setLongitude(testOrder.getEndLon()); destination.setLatitude(testOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(2); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 redisService.setCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + testOrder.getVehicleId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); }else{ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); } } }else{ orderType = OrderType.TEST_CRUISE_ORDER; //巡游单 if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) { /* 构建导航数据给到车端 */ origin.setLongitude(testOrder.getStartLon()); origin.setLatitude(testOrder.getStartLat()); destination.setLongitude(testOrder.getEndLon()); destination.setLatitude(testOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(3); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); } } vehicleId = testOrder.getVehicleId(); }else{ orderType = OrderType.USER_ORDER; //正常单 R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); if (200 != xpOrderR.getCode()) { throw new RuntimeException("订单不存在 " + orderId); } XpOrder xpOrder = xpOrderR.getData(); XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); //驶往上车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) { Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_START_DATA:" + xpOrder.getAppUserId()); if(null == cacheObject){ //查询车辆位置 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()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId)); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(1); dto.setVehicleId(xpOrder.getVehicleId()); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); //缓存数据，用于后续推给乘客导航线路 redisService.setCacheObject("ORDER_NAVIGATION_START_DATA:" + xpOrder.getAppUserId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); }else{ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); }; } //驶往下车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) { //直接取缓存中的导航数据推送给车端和乘客 Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId()); if(null == cacheObject){ //没有缓存数据，重新规划线路 //查询起点到终点的线路 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); dto.setRouteType(2); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 redisService.setCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); }else{ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); } } vehicleId = xpOrder.getVehicleId(); } if(StringUtils.isNotEmpty(routeID)){ log.info("开始推送导航信息 {}", orderId); XpVehicle xpVehicle = xpVehicleClient.queryById(vehicleId).getData(); R stringR = xpMapClient.queryRouted(routeID); if (200 != stringR.getCode()) { throw new RuntimeException("根据导航ID获取车端需要的导航数据失败"); } log.info("根据导航ID获取车端需要的导航数据 订单 {} 导航数据 {}", orderId, stringR.getData()); //构建导航下发请求对象 cloudPushNavigationInfo(requestId, routeID, stringR.getData(), orderType, orderId, xpVehicle.getVehicleVin(), xpVehicle.getXpuId(), taskId); } } /** * 云端订单操作 * 更新途经点/终点 * @param routeVo */ public void cloudOrderOperation(Integer orderType, String orderId, String taskId, RouteVo routeVo, String vin) { //构建消息体 OrderOperation.Builder orderOperationBuilder = OrderOperation.newBuilder(); orderOperationBuilder.setOrderId(orderId); orderOperationBuilder.setTaskId(taskId); orderOperationBuilder.setOperationType(OrderOperationType.CLOUD_UPDATE_WAYPOINTS); //构建航点 List waypoints = new ArrayList<>(); List polyline = routeVo.getPolyline(); for (int i = 0; i < polyline.size(); i++) { LonLatDto lonLatDto = polyline.get(i); Waypoint.Builder waypointBuilder = Waypoint.newBuilder(); waypointBuilder.setWaypointId(UUID.randomUUID().toString()); if (0 == i) { waypointBuilder.setPointType(PointType.START_POINT); //订单类型不同，点位的动作类型不同 waypointBuilder.setActionType(1 == orderType ? ActionType.PICK_UP : ActionType.ACTION_UNSPECIFIED); } else if (polyline.size() - 1 == i) { waypointBuilder.setPointType(PointType.END_POINT); waypointBuilder.setActionType(1 == orderType ? ActionType.DROP_OFF : ActionType.ACTION_UNSPECIFIED); } else { waypointBuilder.setPointType(PointType.WAY_POINT); //订单类型不同，点位的动作类型不同 waypointBuilder.setActionType(ActionType.ACTION_UNSPECIFIED); } //构建经纬度 GeoPoint.Builder geoPointBuilder = GeoPoint.newBuilder(); geoPointBuilder.setLongitude(lonLatDto.getLongitude()); geoPointBuilder.setLatitude(lonLatDto.getLatitude()); waypointBuilder.setPoint(geoPointBuilder.build()); waypointBuilder.setSequence(i + 1); waypoints.add(waypointBuilder.build()); } orderOperationBuilder.addAllWaypoints(waypoints); long sequence = snowflake.nextId(); // 处理数据 redisService.setCacheObject(sequence + "","pushOrder_"+orderId+"_"+vin,30L, TimeUnit.MINUTES); //开发发送请求到XPU xpuUtil.cloudOrderOperation(orderOperationBuilder.build(), vin,sequence); } /** * 构建响应消息 */ public void responseMessage(RTCmdType cmd, Long request_seq, String vin) { ResponseMessage.Builder responseMessageBuilder = ResponseMessage.newBuilder(); responseMessageBuilder.setRequestSeq(request_seq); responseMessageBuilder.setCode(StatusCode.OK); responseMessageBuilder.setTimestamp(System.currentTimeMillis()); xpuUtil.responseMessage(cmd, responseMessageBuilder.build(), vin); } @Resource OrderMissionClient orderMissionClient; /** * 订单状态更新事件 * @param msgContent */ public void orderStateUpdateEvent(RTCmdType cmd, MsgContent msgContent) { //先响应消息 long sequence = msgContent.getHeader().getSequence(); //开始处理消息对应的业务逻辑 OrderStatusInfoEvent orderStatusInfoEvent = msgContent.getOrderStatusInfoEvent(); //只处理请求消息 if (orderStatusInfoEvent.getEventType().equals(RtEventType.kEventOrderStateUpdate)) { OrderStatusInfo orderStatusInfo = orderStatusInfoEvent.getOrderStatusInfo(); String orderId = orderStatusInfo.getOrderId(); //普通单 OrderStatus status = orderStatusInfo.getStatus(); switch (status) { case ORDER_STATUS_UNSPECIFIED: //订单状态未知 log.info("订单状态未知，不做任何处理：{}", orderId); break; case ORDER_RECEIVED: //收到订单 if(!orderId.contains("XU")){ if(orderId.contains("TEST")){ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.READY_DEPART.getCode()); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.READY_DEPART, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.READY_DEPART.getCode(), false); } } break; case ORDER_STARTED: //开始执行订单 log.info("开始执行订单，不做任何处理：{}", orderId); break; case ORDER_EN_ROUTE_TO_PICKUP: //前往上车点 if(!orderId.contains("XU")){ if(orderId.contains("TEST")){ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.PICK_UP.getCode()); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.PICK_UP, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP.getCode(), false); } } break; case ORDER_ARRIVED_AT_PICKUP: //到达上车点 log.info("到达上车点，不做任何处理：{}", orderId); break; case ORDER_WAITING_FOR_PICKUP: //等待乘客上车 if(!orderId.contains("XU")){ if(orderId.contains("TEST")){ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.PICK_UP_GUESTS.getCode()); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.PICK_UP_GUESTS, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode(), false); } } break; case ORDER_PASSENGER_ONBOARD: //先手动暂停触发，因为现目前车机响应状态时机不对 try { Thread.sleep(waittime); } catch (InterruptedException e) { throw new RuntimeException(e); } //乘客已上车 if(!orderId.contains("XU")){ if(orderId.contains("TEST")){ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.PASSENGER_BOARDED.getCode()); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.PASSENGER_BOARDED, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode(), false); } } break; case ORDER_EN_ROUTE_TO_DESTINATION: //前往目的地 if(!orderId.contains("XU")){ if(orderId.contains("TEST")){ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.IN_SERVICE.getCode()); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.IN_SERVICE, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.IN_SERVICE.getCode(), false); } } break; case ORDER_ARRIVED_AT_DESTINATION: //到达目的地 log.info("到达目的地，不做任何处理：{}", orderId); break; case ORDER_WAITING_FOR_DROPOFF: //等待乘客下车，开始倒计时发下巡游单 if(!orderId.contains("XU")){ if(orderId.contains("TEST")){ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.SEE_OFF_GUESTS.getCode()); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), false); } } break; case ORDER_DROPOFF_COMPLETED: //下车完成，不需要处理状态，由后续的乘客主动触发或者定时任务触发 log.info("乘客已下车，不做任何处理：{}", orderId); break; case ORDER_COMPLETED: //已完成 log.info("已完成，不做任何处理：{}", orderId); if(!orderId.contains("XU")){ XpOrder xpOrder = xpOrderClient.queryOrderById(Long.valueOf(orderId)).getData(); if (Objects.nonNull(xpOrder)){ // 查询车辆 XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); if (Objects.nonNull(xpVehicle)){ String vin = xpVehicle.getVehicleVin(); // 封装埋点消息实体 VehicleEventDTO vehicleEventDTO = new VehicleEventDTO(); vehicleEventDTO.setVin(vin); vehicleEventDTO.setXpuId(""); vehicleEventDTO.setOrderId(orderId); vehicleEventDTO.setEventKey("RT_xpu_vehicle_trace_upload"); vehicleEventDTO.setEventTime(System.currentTimeMillis()); Map attributes = new HashMap<>(); attributes.put("vin",vin); attributes.put("order_id",vehicleEventDTO.getOrderId()); attributes.put("order_type","user_order"); attributes.put("timestamp",vehicleEventDTO.getEventTime()); attributes.put("current_main_status",OrderStatus.ORDER_COMPLETED); attributes.put("current_sub_status",""); VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class); if(Objects.nonNull(vehiclePosition)){ attributes.put("odometer_km",vehiclePosition.getTotalOdometer()); }else { attributes.put("odometer_km",""); } vehicleEventDTO.setAttributes(attributes); vehicleBuriedPointClient.sendVehicleEvent(vehicleEventDTO); log.info("已完成，推送埋点：{}", orderId); } } } break; case ORDER_CANCELLED: //已取消 if(!orderId.contains("XU")){ String taskId = orderStatusInfo.getTaskId(); JSONObject reassignCancelContext = resolveReassignCancelContext(orderId, orderStatusInfo); if (Objects.nonNull(reassignCancelContext)) { log.info("改派旧车取消回调，跳过订单取消处理，orderId: {}, taskId: {}, cancelScene: {}, context: {}", orderId, taskId, reassignCancelContext.getString("cancelScene"), reassignCancelContext); //响应消息 String responseVin = resolveReassignCancelResponseVin(reassignCancelContext); if (StringUtils.isNotEmpty(responseVin)) { responseMessage(cmd, sequence, responseVin); } else { log.warn("改派旧车取消回调未找到响应VIN，orderId: {}, taskId: {}, context: {}", orderId, taskId, reassignCancelContext); } return; } if(orderId.contains("TEST")){ //判断订单状态，已经取消的不作处理 Object cacheObject = redisService.getCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4)); if(null != cacheObject){ int order_status = Integer.parseInt(cacheObject.toString()); if(0 == order_status){ return; } } XpTestOrder testOrder = testOrderClient.queryOrderById(Long.valueOf(orderId.substring(4))).getData(); if(testOrder.getStatus() == 0 || testOrder.getStatus() >= 8){ return; } //行程中取消订单，按照行程结束处理 if(testOrder.getStatus() > 5 && testOrder.getStatus() < 8){ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.SEE_OFF_GUESTS.getCode()); redisService.setCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), 24L, TimeUnit.HOURS); }else{ testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.CANCELED.getCode()); redisService.setCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.CANCELED.getCode(), 24L, TimeUnit.HOURS); } }else{ //判断订单状态，已经取消的不作处理 Object cacheObject = redisService.getCacheObject("ORDER_STATUS:" + orderId); if(null != cacheObject){ int order_status = Integer.parseInt(cacheObject.toString()); if(0 == order_status){ return; } } XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData(); if(xpOrder.getStatus() == 0 || xpOrder.getStatus() >= 8){ return; } List xczqx = new ArrayList(){{ add(OrderStatusEnum.IN_SERVICE.getCode());//服务中 add(OrderStatusEnum.SEE_OFF_GUESTS.getCode());//SEE_OFF_GUESTS }}; if(xczqx.contains(xpOrder.getStatus())){//行程中取消订单，按照行程结束处理 buildPushOrderStatus(cmd, orderId, OrderStatusEnum.SEE_OFF_GUESTS, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.SEE_OFF_GUESTS.getCode(), true); redisService.setCacheObject("ORDER_STATUS:" + orderId, OrderStatusEnum.SEE_OFF_GUESTS.getCode(), 24L, TimeUnit.HOURS); // 判断当前订单的车辆是否有连环派单，然后判断是都还满足连环派单的条件，不满足则需要重新派单 Long nextOrderId = redisService.getCacheObject(CacheConstants.nextOrder(orderId)); if(null != nextOrderId){ XpOrder nextOrder = xpOrderClient.queryOrderById(nextOrderId).getData(); if(null != nextOrder && nextOrder.getStatus() == OrderStatusEnum.PICK_UP.getCode()){ redisService.setCacheObject(CacheConstants.nextOrder(orderId),null); ReassignOrderDTO reassignOrderDTO = new ReassignOrderDTO(); reassignOrderDTO.setUid(xpOrder.getAppUserId().toString()); reassignOrderDTO.setOrderNo(xpOrder.getOrderNumber()); reassignOrderDTO.setCount(10); reassignOrderDTO.setLongitude(xpOrder.getStartLon()); reassignOrderDTO.setLatitude(xpOrder.getStartLon()); reassignOrderClient.reassignOrder(reassignOrderDTO); } } redisService.getCacheObject("ORDER_CONNECT_ORDER:" + orderId); }else{ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.CANCELED, sequence); xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.CANCELED.getCode(), true); redisService.setCacheObject("ORDER_STATUS:" + orderId, OrderStatusEnum.CANCELED.getCode(), 24L, TimeUnit.HOURS); } } } break; case ORDER_ARRIVED_AT_CRUISE_DESTINATION: //到达巡游终点 if(orderId.contains("XU")){ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.COMPLETED, sequence); String cruiseOrderId = orderId.substring(2); XpCruiseOrder xpCruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(cruiseOrderId)).getData(); if (Objects.nonNull(xpCruiseOrder)){ // 查询车辆 XpVehicle xpVehicle = xpVehicleClient.queryById(xpCruiseOrder.getVehicleId()).getData(); if (Objects.nonNull(xpVehicle)){ String vin = xpVehicle.getVehicleVin(); // 封装埋点消息实体 VehicleEventDTO vehicleEventDTO = new VehicleEventDTO(); vehicleEventDTO.setVin(vin); vehicleEventDTO.setXpuId(""); vehicleEventDTO.setOrderId(cruiseOrderId); vehicleEventDTO.setEventKey("RT_xpu_vehicle_trace_upload"); vehicleEventDTO.setEventTime(System.currentTimeMillis()); Map attributes = new HashMap<>(); attributes.put("vin",vin); attributes.put("order_id",vehicleEventDTO.getOrderId()); attributes.put("order_type","cruise_order"); attributes.put("timestamp",vehicleEventDTO.getEventTime()); attributes.put("current_main_status",OrderStatus.ORDER_ARRIVED_AT_CRUISE_DESTINATION); attributes.put("current_sub_status",""); VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class); if(Objects.nonNull(vehiclePosition)){ attributes.put("odometer_km",vehiclePosition.getTotalOdometer()); }else { attributes.put("odometer_km",""); } vehicleEventDTO.setAttributes(attributes); vehicleBuriedPointClient.sendVehicleEvent(vehicleEventDTO); log.info("到达巡游终点，推送埋点：{}", orderId); } } }else if(orderId.contains("TEST")){ //测试巡游单完成状态 XpTestOrder testOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4))).getData(); if(testOrder.getOrderType() == 2){ XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData(); //响应消息 responseMessage(cmd, sequence, xpVehicle.getVehicleVin()); //测试巡游单处理逻辑 testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.COMPLETED.getCode()); redisService.setCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.COMPLETED.getCode(), 24L, TimeUnit.HOURS); } } break; case ORDER_PAUSED: //订单暂停 log.info("订单暂停，不做任何处理：{}", orderId); break; default: break; } } } /** * 构建订单状态推送消息 * @param orderId * @param status */ public void buildPushOrderStatus(RTCmdType cmd, String orderId, OrderStatusEnum status, Long sequence){ if(orderId.contains("XU")){ //巡游单 //到达目的地 if(OrderStatusEnum.COMPLETED.equals(status)){ // 返程找到目前的订单 -》之前的订单 查询生成一个 正常巡游 找到下一个点位 若到达最后点位 则反向找点位 每次判断是否收车状态 是则不进行处理了 // 找到这个订单 R cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2))); if(cruiseOrder!=null){ XpCruiseOrder xpCruiseOrder = cruiseOrder.getData(); Long vehicleId = xpCruiseOrder.getVehicleId(); // 完成当前巡游单子 已完成 xpCruiseOrderClient.completeCruiseOrder(Long.parseLong(orderId.substring(2))); R xpVehicleR = xpVehicleClient.queryById(vehicleId); XpVehicle xpVehicle = xpVehicleR.getData(); //响应消息 responseMessage(cmd, sequence, xpVehicle.getVehicleVin()); // 运营中 继续执行 if(xpVehicle.getOperatingStatus()==1 && xpVehicle.getState()==0){ log.info("巡游单到达终点，继续执行下一笔巡游单"); // 创建新的巡游单子 判断是否为返程单子 // 正常巡游 找到下一个点位 若到达最后点位 则反向找点位 // 起点 Double endLon = xpCruiseOrder.getEndLon(); Double endLat = xpCruiseOrder.getEndLat(); XpCruiseOrder newXpCruiseOrder = new XpCruiseOrder(); newXpCruiseOrder.setVehicleId(xpCruiseOrder.getVehicleId()); newXpCruiseOrder.setVehicleVin(xpCruiseOrder.getVehicleVin()); newXpCruiseOrder.setFleetId(xpCruiseOrder.getFleetId()); newXpCruiseOrder.setOrderType(1); newXpCruiseOrder.setStartLat(endLat); newXpCruiseOrder.setStartLon(endLon); newXpCruiseOrder.setStartPointId(xpCruiseOrder.getEndPointId()); newXpCruiseOrder.setStartPointName(xpCruiseOrder.getEndPointName()); newXpCruiseOrder.setStatus(1); newXpCruiseOrder.setLastOrderId(xpCruiseOrder.getId()); // 通过车找终点id CruisePointDto cruisePointDto = new CruisePointDto(); cruisePointDto.setNewXpCruiseOrder(newXpCruiseOrder); cruisePointDto.setVehicleId(xpCruiseOrder.getVehicleId()); cruisePointDto.setPointId(newXpCruiseOrder.getStartPointId()); cruisePointDto.setOrderReverse(xpCruiseOrder.getOrderReverse()); R cruisePoint = xpCruisePlanClient.getCruisePoint(cruisePointDto); if(cruisePoint!=null){ xpCruiseOrderClient.addCruiseOrder(cruisePoint.getData()); } } } } }else{ //普通单 R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); XpOrder xpOrder = xpOrderR.getData(); XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); //响应消息 responseMessage(cmd, sequence, xpVehicle.getVehicleVin()); //构建推送消息 Map params = new HashMap<>(); Map content = new HashMap<>(); content.put("orderId", orderId.toString()); content.put("status", status.getCode()); params.put("type", PushMessageTypeEnum.ORDER_STATUS.getDesc()); params.put("content", JSONObject.toJSONString(content)); params.put("timestamp", System.currentTimeMillis() + ""); String message = JSONObject.toJSONString(params); Long appUserId = xpOrder.getAppUserId(); SendMessageDTO dto = new SendMessageDTO(); dto.setAppUserId(appUserId + ""); dto.setMessage(message); R stringR = xpPushMessageClient.sendStatus(dto); //推送失败，重试3次 if(stringR.getCode() != Constants.SUCCESS){ ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()); executor.execute(() -> { for (int i = 0; i < 3; i++) { R stringR1 = xpPushMessageClient.sendStatus(dto); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("推送订单状态消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(dto), xpOrder.getOrderNumber()); } }); } } } private JSONObject getReassignCancelContext(String orderId) { if (StringUtils.isEmpty(orderId)) { return null; } Object cacheObject = redisService.getCacheObject(REASSIGN_CANCEL_CONTEXT_PREFIX + orderId); if (cacheObject == null) { return null; } return JSON.parseObject(cacheObject.toString()); } private JSONObject resolveReassignCancelContext(String orderId, OrderStatusInfo orderStatusInfo) { if (OrderStateChangeSource.ORDER_STATE_CHANGE_SOURCE_VEHICLE != orderStatusInfo.getChangeSource()) { return null; } JSONObject cancelContext = getReassignCancelContext(orderId); if (cancelContext == null) { return null; } if (CancelOrderSceneEnum.REASSIGN_OLD_VEHICLE != CancelOrderSceneEnum.fromCode( cancelContext.getString("cancelScene"))) { return null; } R orderResult = xpOrderClient.queryOrderById(Long.parseLong(orderId)); XpOrder xpOrder = orderResult == null ? null : orderResult.getData(); if (xpOrder == null) { log.warn("改派取消回调识别失败，订单不存在，orderId: {}, context: {}", orderId, cancelContext); return null; } Long newVehicleId = cancelContext.getLong("newVehicleId"); Integer currentStatus = xpOrder.getStatus(); boolean vehicleMatched = Objects.nonNull(newVehicleId) && Objects.equals(newVehicleId, xpOrder.getVehicleId()); boolean statusMatched = Objects.nonNull(currentStatus) && currentStatus > 0 && currentStatus < 8; if (vehicleMatched && statusMatched) { return cancelContext; } log.info("改派取消回调未命中专用上下文，orderId: {}, currentVehicleId: {}, newVehicleId: {}, currentStatus: {}", orderId, xpOrder.getVehicleId(), newVehicleId, currentStatus); return null; } private String resolveReassignCancelResponseVin(JSONObject cancelContext) { if (cancelContext == null) { return null; } String oldVehicleVin = cancelContext.getString("oldVehicleVin"); if (StringUtils.isNotEmpty(oldVehicleVin)) { return oldVehicleVin; } Long oldVehicleId = cancelContext.getLong("oldVehicleId"); if (Objects.isNull(oldVehicleId)) { return null; } R vehicleResult = xpVehicleClient.queryById(oldVehicleId); XpVehicle xpVehicle = vehicleResult == null ? null : vehicleResult.getData(); return xpVehicle == null ? null : xpVehicle.getVehicleVin(); } /** * 云端下发导航到XPU * @param requestId XPU请求id * @param routeId 高德导航id * @param routeDataContent 高德导航原数据 * @param orderId 订单id * @param vin 车辆VIN * @param xpuId XPU设备id */ public void cloudPushNavigationInfo(String requestId, String routeId, String routeDataContent, String orderType, String orderId, String vin, String xpuId, String taskId) { //组装报文，将routeDateContent信息推送到XPU RestoreNavigationData.Builder navigationDataBuilder = RestoreNavigationData.newBuilder(); navigationDataBuilder.setRouteId(routeId); navigationDataBuilder.setRequestId(requestId); navigationDataBuilder.setContent(routeDataContent); navigationDataBuilder.setTimestamp(System.currentTimeMillis()); navigationDataBuilder.setOrderId(orderId); navigationDataBuilder.setTaskId(taskId); long sequence = snowflake.nextId(); log.info("开始推送导航数据 {}", orderId); MqttPublishDto mqttPublishDto = xpuUtil.cloudPushNavigation(navigationDataBuilder.build(), xpuId, sequence); log.info("推送导航的响应消息，{}", JSON.toJSONString(mqttPublishDto)); } }