package com.xiaopeng.automaticPilot.service; import cn.hutool.core.lang.Snowflake; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.google.common.collect.ImmutableMap; import com.google.protobuf.InvalidProtocolBufferException; import com.google.protobuf.util.JsonFormat; import com.xiaopeng.api.dto.automaticPilot.MqttPublishDto; import com.xiaopeng.api.dto.automaticPilot.PushOrderDto; import com.xiaopeng.api.dto.automaticPilot.VehicleEventDTO; import com.xiaopeng.api.dto.map.OrderRouteAndFareDto; import com.xiaopeng.api.dto.map.RouteAndFareDto; 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.*; import com.xiaopeng.api.feignClient.order.XpCruiseOrderClient; import com.xiaopeng.api.feignClient.order.XpOrderClient; import com.xiaopeng.api.feignClient.order.XpOrderReassignClient; import com.xiaopeng.api.feignClient.order.XpTestOrderClient; import com.xiaopeng.api.feignClient.pushMessage.WebSocketMessageClient; 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.operation.XpSystemSetting; import com.xiaopeng.api.model.operation.XpTestPlanDemoOrder; import com.xiaopeng.api.model.order.*; 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.pushMessage.UpdateEndAddStatusVo; import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo; import com.xiaopeng.automaticPilot.job.base.QuartzManager; import com.xiaopeng.automaticPilot.job.base.TimeJobType; import com.xiaopeng.automaticPilot.job.jobs.OrderExceptionJob; 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.DateUtils; 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.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.time.LocalDateTime; 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:"; private static final String CANCEL_ORDER_CONTEXT_PREFIX = "CANCEL_ORDER_CONTEXT:"; // private static final String ORDER_DROPOFF_CONFIRMED_PREFIX = "ORDER_DROPOFF_CONFIRMED:"; // private static final String ORDER_DROPOFF_CRUISE_DISPATCHED_PREFIX = "ORDER_DROPOFF_CRUISE_DISPATCHED:"; private static final String PICKUP_EXCEPTION_FREE_CANCEL_PREFIX = "PICKUP_EXCEPTION_FREE_CANCEL:"; private static final String PICKUP_EXCEPTION_LIVE_LOCATION_PREFIX = "PICKUP_EXCEPTION_LIVE_LOCATION:"; private static final String PICKUP_EXCEPTION_HANDLED_PREFIX = "PICKUP_EXCEPTION_HANDLED:"; private static final double PICKUP_EXCEPTION_MIN_DISTANCE_METERS = 30D; private static final double PICKUP_EXCEPTION_MAX_DISTANCE_METERS = 200D; private static final int PICKUP_EXCEPTION_TYPE_REMOTE_STOP = 1; private static final int PICKUP_EXCEPTION_FREE_CANCEL_EXTEND_MINUTES = 10; private static final int PICKUP_EXCEPTION_LIVE_LOCATION_MINUTES = 15; private static final String PICKUP_EXCEPTION_MESSAGE = "为确保行车安全，已为您调整新的上车点，请步行导航至车辆位置"; @Resource private XpMapClient xpMapClient; @Resource private XpOrderClient xpOrderClient; @Resource private OrderMissionClient orderMissionClient; @Resource private XpCruiseOrderClient xpCruiseOrderClient; @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 RedisService redisService; @Resource private XpCruisePlanClient xpCruisePlanClient; @Resource private XpOrderReassignClient xpOrderReassignClient; @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; @Resource private WebSocketMessageClient webSocketMessageClient; @Resource private XpSystemSettingClient systemSettingClient; @Resource private XpTestPlanClient xpTestPlanClient; @Resource private XpTestPlanDemoClient xpTestPlanDemoClient; private static final JsonFormat.Printer jsonPrinter = JsonFormat.printer() .printingEnumsAsInts() .preservingProtoFieldNames(); private static final String TEST_PLAN_DEMO_ORDER_PREFIX = "TPD"; /** * 消费消息 - XPU上报Robotaxi云端的消息topic（高优先级） * */ public void topicProdRobotaxiXpuHigh(String message) { MsgContent msgContent = parseMsgContent(message); if (msgContent != null) { String topic = JSON.parseObject(message).getString("topic"); String xpuId = topic.substring(topic.lastIndexOf("/") + 1); handleXPUMessage(msgContent, message, xpuId); } } /** * 消费消息 - XPU上报Robotaxi云端的消息topic（低优先级） */ public void topicProdRobotaxiXpuNormal(String message) { MsgContent msgContent = parseMsgContent(message); if (msgContent != null) { String topic = JSON.parseObject(message).getString("topic"); String xpuId = topic.substring(topic.lastIndexOf("/") + 1); handleXPUMessage(msgContent, message, xpuId); } } 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, String xpuId){ 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, xpuId); break; case TRIP_STATUS_UPDATE_EVENT: log.info("行程状态更新(XPU→云端)(XPU→座舱)：{}", message); break; case PULLOVER_STATUS_EVENT: log.info("靠边停车状态(XPU→云端)(XPU→座舱)：{}", message); pulloverStatusEvent(msgContent); 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); orderNearDestinationEvent(cmd, msgContent); break; case USER_OPERATION_INFO_EVENT: log.info("用户操作事件 (XPU→云端)：{}", message); userOperationInfoEvent(cmd, msgContent); break; default: log.info("消息不在处理范围内，忽略：{}", message); } } //消息应答 if(MessageType.MESSAGE_TYPE_RESPONSE == type) { //区分业务类型 int cmdValue = header.getCmd(); RTCmdType cmd = RTCmdType.getByValue(cmdValue); if (cmd == null) { log.info("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.info("订单操作失败，orderId {} taskId {} 错误码 {}", rtTaskResult.getOrderId(), rtTaskResult.getTaskId(), rtTaskResult.getStatus()); // 车端返回非 ACCEPTED，除记录日志外还需对用户单进行改派处理。 // 参考 popOrder 失败的处理逻辑：回滚车辆状态、给车端发送取消任务、车辆继续巡游并重新派单。 // 仅处理用户单（user_order），其他类型（巡游单、测试单等）保持原有仅记录日志的行为。 Object failSequence = redisService.getCacheObject("CREATE_ORDER:" + rtTaskResult.getOrderId()); if (failSequence != null) { JSONObject failJsonObject = JSON.parseObject(failSequence.toString()); String failOrderType = failJsonObject.getString("orderType"); if (OrderType.USER_ORDER.equals(failOrderType)) { String failOrderId = failJsonObject.getString("orderId"); String failVin = failJsonObject.getString("vin"); // 幂等：同一用户单+车辆的重复响应只处理一次。两条响应几乎同时到达， // 用 setIfAbsent 原子抢占，抢不到说明已被处理/正在处理，直接忽略，避免重复改派、重复派单。 if (!Boolean.TRUE.equals(redisService.setIfAbsent("RT_TASK_HANDLED:" + failOrderId + ":" + failVin, "1", 30L, TimeUnit.MINUTES))) { log.warn("CLOUD_SEND_RT_TASK 重复响应（非ACCEPTED），已处理/处理中，忽略。orderId={} vin={}", failOrderId, failVin); return; } XpOrder xpOrder = xpOrderClient.queryOrderById(Long.valueOf(failOrderId)).getData(); //车辆状态回执为空闲 XpVehicle xpVehicle = xpVehicleClient.findByVehicleVin(failVin).getData(); XpVehicle xpVehicle1 = new XpVehicle(); xpVehicle1.setId(xpVehicle.getId()); xpVehicle1.setState(0); xpVehicleClient.updateVehicleById(xpVehicle1); log.info("车端返回非ACCEPTED，车辆状态修改0 {} {}", xpVehicle.getPlateNumber(), xpOrder.getId()); //业务执行失败，给车端发送取消任务报文，继续执行巡游任务 PushOrderDto pushOrderDto = new PushOrderDto(); pushOrderDto.setEndTaskId(xpOrder.getXpuStartTaskId()); pushOrderDto.setOrderId(failOrderId); pushOrderDto.setVin(failVin); pushOrderDto.setOperation_type(OrderOperationType.CLOUD_CANCEL_ORDER); orderMissionClient.cancelOrder(pushOrderDto); //车辆继续巡游 xpOrderClient.addNewCruiseMission(failVin, failOrderId); log.error("车端返回非ACCEPTED，用户单改派处理，车辆继续执行巡游 {}", failOrderId); //将当前车辆添加到禁推列表中 Object cacheObject1 = redisService.getCacheObject("CAN_NOT_RECEIVE_ORDER_VEHICLE:" + xpOrder.getId()); List vinList = new ArrayList<>(); if(Objects.nonNull(cacheObject1)){ vinList = JSON.parseArray(cacheObject1.toString(), String.class); } vinList.add(xpVehicle.getVehicleVin()); redisService.setCacheObject("CAN_NOT_RECEIVE_ORDER_VEHICLE:" + xpOrder.getId(), JSON.toJSONString(vinList), 10L, TimeUnit.MINUTES); //放入队列，重新执行派单 xpOrderClient.againPushOrder(failOrderId); } } return; } Object sequence = redisService.getCacheObject("CREATE_ORDER:" + response.getResults(0).getOrderId()); if(sequence!=null){ JSONObject jsonObject = JSON.parseObject(sequence.toString()); // 校验响应回执的 taskId 是否属于本次云端下发的 taskId 集合 List cachedTaskIds = jsonObject.getList("taskIds", String.class); if (cachedTaskIds != null && !cachedTaskIds.isEmpty() && !cachedTaskIds.contains(rtTaskResult.getTaskId())) { log.warn("CLOUD_SEND_RT_TASK 响应中的 taskId 不在本次下发集合中，忽略处理。orderId={} 响应 taskId={} 已下发 taskIds={}", rtTaskResult.getOrderId(), rtTaskResult.getTaskId(), cachedTaskIds); return; } String orderId = jsonObject.getString("orderId"); String vin = jsonObject.getString("vin"); String orderType = jsonObject.getString("orderType"); // 幂等：同一用户单+车辆的重复响应只处理一次。两条响应几乎同时到达， // 用 setIfAbsent 原子抢占，抢不到说明已被处理/正在处理，直接忽略，避免重复摘单、重复推送。 if (!Boolean.TRUE.equals(redisService.setIfAbsent("RT_TASK_HANDLED:" + orderId + ":" + vin, "1", 30L, TimeUnit.MINUTES))) { log.warn("CLOUD_SEND_RT_TASK 重复响应（ACCEPTED），已处理/处理中，忽略。orderId={} vin={}", orderId, vin); return; } PopOrderDto dto = new PopOrderDto(); if (isTestPlanDemoOrder(orderId)) { dto.setOrderId(parseTestPlanDemoOrderId(orderId)); } else 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); R r = xpOrderClient.popOrder(dto); if(200 != r.getCode()){ if (isTestPlanDemoOrder(orderId)) { // 与用户单一致：摘单失败时回滚车辆状态并给车端发送取消任务， // 避免车辆卡在“接单中”状态。 // TODO 续巡游单：TPD 暂不参与巡游续单（不调用 addNewCruiseMission/againPushOrder）， // 待后续需求确认是否需要补充重派/续单逻辑。 log.error("handle test plan demo order pop response failed {}", orderId); XpVehicle demoVehicle = xpVehicleClient.findByVehicleVin(vin).getData(); if (demoVehicle != null) { demoVehicle.setState(0); xpVehicleClient.updateVehicleById(demoVehicle); } PushOrderDto demoCancelDto = new PushOrderDto(); demoCancelDto.setOrderId(orderId); demoCancelDto.setVin(vin); demoCancelDto.setOperation_type(OrderOperationType.CLOUD_CANCEL_ORDER); orderMissionClient.cancelOrder(demoCancelDto); log.info("测试Demo订单摘单失败，已回滚车辆状态并给车端发送取消任务 {}", orderId); return; } XpOrder xpOrder = xpOrderClient.queryOrderById(Long.valueOf(orderId)).getData(); //车辆状态回执未空闲 XpVehicle xpVehicle = xpVehicleClient.findByVehicleVin(vin).getData(); XpVehicle xpVehicle1 = new XpVehicle(); xpVehicle1.setId(xpVehicle.getId()); xpVehicle1.setState(0); xpVehicleClient.updateVehicleById(xpVehicle1); log.info("处理车辆接单后的业务逻辑异常，车辆状态修改0 {} {}", xpVehicle.getPlateNumber(), orderId); //业务执行失败，给车端发送取消任务报文，继续执行巡游任务 PushOrderDto pushOrderDto = new PushOrderDto(); pushOrderDto.setEndTaskId(xpOrder.getXpuStartTaskId()); pushOrderDto.setOrderId(orderId); pushOrderDto.setVin(vin); pushOrderDto.setOperation_type(OrderOperationType.CLOUD_CANCEL_ORDER); orderMissionClient.cancelOrder(pushOrderDto); log.info("处理车辆接单后的业务逻辑异常，开始执行巡游订单 {}", orderId); //车辆继续巡游 xpOrderClient.addNewCruiseMission(vin, orderId); log.error("处理车辆接单后的业务逻辑异常，车辆继续执行巡游 {}", orderId); //将当前车辆添加到禁推列表中 Object cacheObject1 = redisService.getCacheObject("CAN_NOT_RECEIVE_ORDER_VEHICLE:" + xpOrder.getId()); List vinList = new ArrayList<>(); if(Objects.nonNull(cacheObject1)){ vinList = JSON.parseArray(cacheObject1.toString(), String.class); } vinList.add(xpVehicle.getVehicleVin()); redisService.setCacheObject("CAN_NOT_RECEIVE_ORDER_VEHICLE:" + xpOrder.getId(), JSON.toJSONString(vinList), 10L, TimeUnit.MINUTES); //放入队列，重新执行派单 xpOrderClient.againPushOrder(orderId); return; } 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 = null; if(null != runningOrderStatusInfo){ order_id = runningOrderStatusInfo.getOrderId(); } // 仅在存在订单时才需要订单级分布式锁。当 order_id 为空时，锁 key 会退化为全局唯一的 // "PUSH_ORDER_ROUTE_LOCK:"，导致所有无订单车辆的高频位置上报争抢同一把锁而被串行化， // 是消息堆积的主要原因。空订单仅做位置存储、不涉及订单线路推送，因此无需加锁。 boolean needLock = StringUtils.isNotEmpty(order_id); RLock lock = needLock ? redissonClient.getLock("PUSH_ORDER_ROUTE_LOCK:" + order_id) : null; boolean locked = false; try { if (needLock) { locked = lock.tryLock(10, 10, TimeUnit.SECONDS); if (!locked) { log.info("获取锁失败，orderId：{} 无法完成车辆实时位置的逻辑处理。", 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 if (isTestPlanDemoOrder(order_id)) { orderId = parseTestPlanDemoOrderId(order_id); }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); try { redisService.setCacheObject("VEHICLE_POSITION:" + vin, longitude + "," + latitude); } catch (Exception e) { // 仅记录指标后原样抛出，严格还原埋点前行为： // Redis 写失败时跳过 Mongo 写入与下游乘客推送（由外层 catch 统一兜底） throw e; } try { 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); } vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class); redisService.setCacheObject("VEHICLE_SYSTEM_DATA:" + vin, JSON.toJSONString(vehiclePosition)); } catch (Exception e) { throw e; } //调用推送服务接口，给乘客推送剩余轨迹和里程时长 RtTripInfo rtTripInfo = systemRuntimeStatus.getRunningOrderStatusInfo().getRtTripInfo(); if (null != orderId && !order_id.contains("XU")) { if (isTestPlanDemoOrder(order_id)) { return; } boolean hasNaviProgressInfo = rtTripInfo.hasCurrentTask() && rtTripInfo.getCurrentTask().hasNaviProgressInfo(); if (StringUtils.isNotEmpty(order_id) && !order_id.contains("TEST")) { if (hasNaviProgressInfo) { clearPickupExceptionLiveLocation(order_id); } else { pushPickupExceptionLiveLocation(order_id, longitude, latitude); } } PushOrderRouteDTO dto = new PushOrderRouteDTO(); dto.setMessageType(PushMessageTypeEnum.ORDER_ROUTE.getDesc()); dto.setOrderId(orderId); dto.setRouteType(1); dto.setLongitude(longitude); dto.setLatitude(latitude); dto.setSpeed(dynamicStatus.getSpeed()); if (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(StringUtils.isNotEmpty(order_id) && !order_id.contains("TEST")){ //推送修改目的地权限状态 pushUpdateEndAddressStatus(orderId); xpPushMessageClient.pushOrderRoute(dto);//存储历史轨迹 // 给连环派单的用户推送 String json = redisService.getCacheObject(CacheConstants.nextOrder(order_id)); if (StringUtils.isNotEmpty(json)){ JSONObject jsonObject = JSON.parseObject(json); dto.setOrderId(jsonObject.getLong("orderId")); //判断上一单任然在服务中才需要推送上一单的终点经纬度 Object cacheObject = redisService.getCacheObject("ORDER_STATUS:" + orderId); if(null != cacheObject && !Arrays.asList("0", "1", "8", "9", "10").contains(cacheObject.toString())){ dto.setEndLongitude(jsonObject.getDouble("endLongitude")); dto.setEndLatitude(jsonObject.getDouble("endLatitude")); } xpPushMessageClient.pushOrderRoute(dto); } xpOrderClient.saveVehicleLocation(orderTrajectory); }else{ testOrderClient.saveVehicleLocation(orderTrajectory); } } } }catch (Exception e){ log.error("处理XPU车辆实时位置异常，订单ID：{}", order_id, e); }finally { // 仅在成功获取且当前线程持有锁时释放，避免 tryLock 失败后 unlock 抛 IllegalMonitorStateException if (locked && lock != null && lock.isHeldByCurrentThread()) { lock.unlock(); } } } /** * 推送修改目的地权限状态 * @param orderId */ private void pushUpdateEndAddressStatus(Long orderId){ XpOrder xpOrder = xpOrderClient.queryOrderById(orderId).getData(); Boolean hasKey = redisService.hasKey("UPDATE_END_ADDRESS_STATUS_MSG:" + xpOrder.getAppUserId()); if(hasKey){ return; } XpSystemSetting xpSystemSetting = systemSettingClient.getSystemSettingOne().getData(); //行程中修改目地次数 Integer xczCaCount = xpSystemSetting.getXczCaCount(); Integer status = 0; if(xpOrder.getStatus() < 6 && xpOrder.getOrderChange() <= xczCaCount){ log.info("修改目的地，接驾中允许修改终点 {}", orderId); status = 1; } if(xpOrder.getStatus() == 6){ log.info("修改目的地，开始计算规则是否满足修改终点 {}", orderId); //行程中剩余里程 Integer xczSyDistance = xpSystemSetting.getXczSyDistance(); //行程中剩余时间 Integer xczSyTime = xpSystemSetting.getXczSyTime(); //行程中豁免时速 km/h Integer xczHmSpeedPerHour = xpSystemSetting.getXczHmSpeedPerHour(); //行程中豁免持续时间 分钟 Integer xczHmDuration = xpSystemSetting.getXczHmDuration(); Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId()); if(null != cacheObject){ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); //剩余时间 秒 Integer duration = routeVo.getDuration(); //剩余距离 米 Integer distance = routeVo.getDistance(); if(xczSyDistance < distance && xczSyTime * 60 < duration && xpOrder.getOrderChange() < xczCaCount){ log.info("修改目的地，允许修改目的地 {} 剩余里程 {} 剩余时间", orderId, distance, duration); status = 1; }else{ log.info("修改目的地，不满足修改条件，开始计算豁免 {}", orderId); //计算豁免 //计算平均时速小于设定值，可豁免限制 LocalDateTime localDateTime = LocalDateTime.now().minusMinutes(xczHmDuration); List orderTrajectories = mongoTemplate.find(Query.query(Criteria.where("orderId").is(orderId).and("createTime").gt(localDateTime)), OrderTrajectory.class); log.info("修改目的地，最近持续时间轨迹 {} {}", orderId, JSON.toJSONString(orderTrajectories)); double speed = orderTrajectories.stream().map(OrderTrajectory::getSpeed).reduce(Double::sum).orElse(0D) / orderTrajectories.size(); log.info("修改目的地，最近持续时间平均时速 {} {} km", orderId, speed); if(speed <= xczHmSpeedPerHour){ log.info("修改目的地，满足豁免条件 {} {} km", orderId, xczHmSpeedPerHour); status = 1; } } } } UpdateEndAddStatusVo vo = new UpdateEndAddStatusVo(); vo.setOrderId(xpOrder.getId()); vo.setAppUserId(xpOrder.getAppUserId()); JSONObject message = new JSONObject(); message.put("type", "UPDATE_END_ADDRESS_STATUS"); JSONObject content = new JSONObject(); content.put("status", status); message.put("content", content); message.put("timestamp", System.currentTimeMillis()); vo.setMessage(message.toString()); webSocketMessageClient.sendUpdateEndAddStatus(vo); } /** * 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 (isTestPlanDemoOrder(orderId)) { updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.CANCELED); return; } 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()); // 解绑用户与车辆关系 String vin = vehicle.getVehicleVin(); CancelTaxiPassengerVo cancelTaxiPassengerVo = new CancelTaxiPassengerVo(); cancelTaxiPassengerVo.setVin(vin); 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())){//在接驾中 车辆出现故障 重新派单 改派 //重新放入队列中进行派单 xpOrderClient.reassignmentPushOrder(orderId); log.info("车辆异常，触发订单改派 {}", orderId); } } //添加车辆异常任务和发送消息 xpTaskClient.saveVehicleAccidentError(vehicle.getId()); String vin = vehicle.getVehicleVin(); // 封装埋点消息实体 VehicleEventDTO vehicleEventDTO = new VehicleEventDTO(); vehicleEventDTO.setVin(vin); vehicleEventDTO.setXpuId(""); vehicleEventDTO.setOrderId(""); vehicleEventDTO.setEventKey("RT_web_vehicle_status_changed"); vehicleEventDTO.setEventTime(System.currentTimeMillis()); Map attributes = new HashMap<>(); attributes.put("vin", vin); 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, String xpuId) { //开始处理消息对应的业务逻辑 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, xpuId); break; case ROUTE_TO_DESTINATION: log.info("驶往下车点的行程 {}", orderId); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence(), taskId, xpuId); break; case ROUTE_TO_CRUISE: log.info("巡游阶段路线 {}", orderId); buildNaviRequest(requestId, orderId, orderPhase, msgContent.getHeader().getSequence(), taskId, xpuId); 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 = ""; Float speed = 0f; //添加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; } Double distance = 0.0; Double duration = 0.0; Double drivingDistance = 0.0; if (isTestPlanDemoOrder(orderId)) { // TPD 测试单：参考用户单逻辑做偏航重规划并下发车端（无乘客端推送）。 // 坐标来源于点位表（operation 服务在 getOrderById 中填充）；停车单按订单状态区分接驾/送驾段， // 不停车单仅有送驾段，始终重规划到终点。 orderType = OrderType.TEST_PLAN_DEMO_ORDER; R demoOrderR = xpTestPlanDemoClient.getOrderById(parseTestPlanDemoOrderId(orderId)); XpTestPlanDemoOrder demoOrder = demoOrderR == null ? null : demoOrderR.getData(); if (demoOrder == null || demoOrder.getVehicleId() == null) { log.warn("TPD 偏航重规划跳过，订单或车辆不存在 orderId={}", orderId); return; } Integer demoStatus = demoOrder.getOrderStatus(); // 按订单状态区分接驾/送驾段（与用户单一致）：未到“待客上车”(4)之前视为接驾段。 // 不停车单为单段送驾，状态会很快推进到送驾段，故按状态判断即可，无需读取计划停车开关。 boolean toPickup = demoStatus != null && demoStatus < OrderStatusEnum.PICK_UP_GUESTS.getCode(); origin.setLongitude(longitude); origin.setLatitude(latitude); if (toPickup) { //偏航点 -> 起点（接驾段） destination.setLongitude(demoOrder.getStartLon()); destination.setLatitude(demoOrder.getStartLat()); } else { //偏航点 -> 终点（送驾段） destination.setLongitude(demoOrder.getEndLon()); destination.setLatitude(demoOrder.getEndLat()); } OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(parseTestPlanDemoOrderId(orderId)); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(toPickup ? 1 : 2); dto.setVehicleId(demoOrder.getVehicleId()); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); vehicleId = demoOrder.getVehicleId(); } else 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(); XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData(); String vin = xpVehicle.getVehicleVin(); VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(vin)), VehiclePosition.class); if(null != vehiclePosition){ speed = vehiclePosition.getSpeed(); } //区分测试单和测试巡游单 if(1 == testOrder.getOrderType()){ orderType = OrderType.TEST_ORDER; //驶往上车点的行程 if (4 > 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(); }else { 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(); String vin = xpVehicle.getVehicleVin(); VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(vin)), VehiclePosition.class); if(null != vehiclePosition){ speed = vehiclePosition.getSpeed(); } //驶往上车点的行程 if (xpOrder.getStatus() < 5) { //查询司机当前位置到起点的线路 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.getId(); redisService.setCacheObject(key, JSON.toJSONString(routeVo)); }else { 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(); redisService.setCacheObject(key, JSON.toJSONString(routeVo)); } vehicleId = xpOrder.getVehicleId(); } //直接响应回最新导航数据 if(StringUtils.isNotEmpty(routeID)){ XpVehicle xpVehicle = xpVehicleClient.queryById(vehicleId).getData(); R stringR = xpMapClient.queryRouted(routeID); if (200 != stringR.getCode()) { throw new RuntimeException("根据导航ID获取车端需要的导航数据失败"); } log.info("开始下发XPU导航偏离后新的导航数据：" + orderId); //构建响应导航数据给车端 String vin = xpVehicle.getVehicleVin(); //推送小程序新的导航数据 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); dto.setSpeed(speed); VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(vin)), VehiclePosition.class); dto.setDirection(null != vehiclePosition ? vehiclePosition.getCourse().intValue() : 0); dto.setDuration(routeVo.getDuration().doubleValue()); dto.setDistance(routeVo.getDistance().doubleValue()); dto.setMessageType(PushMessageTypeEnum.ORDER_UPDATE_LINE.getDesc()); xpPushMessageClient.pushOrderRoute(dto); //判断是否有连环单，如果有则需要重新规划连环单的轨迹 String json = redisService.getCacheObject(CacheConstants.nextOrder(orderId)); if (StringUtils.isNotEmpty(json)) { JSONObject jsonObject = JSON.parseObject(json); Long nextOrderId = jsonObject.getLong("orderId"); //主动推送乘客端导航更新了 PushOrderRouteDTO dto2 = new PushOrderRouteDTO(); dto2.setOrderId(nextOrderId); dto2.setRouteType(1); dto2.setLongitude(longitude); dto2.setLatitude(latitude); dto2.setSpeed(speed); //这里使用上一单的剩余距离和时间，模拟车端发的定位数据，后续由推送服务处理真实的距离和时间 dto2.setDirection(null != vehiclePosition ? vehiclePosition.getCourse().intValue() : 0); dto2.setDuration(routeVo.getDuration().doubleValue()); dto2.setDistance(routeVo.getDistance().doubleValue()); dto2.setMessageType(PushMessageTypeEnum.ORDER_UPDATE_LINE.getDesc()); xpPushMessageClient.pushOrderRoute(dto2); } log.info("完成小程序下发导航偏离后的新轨迹：" + orderId); } cloudPushNavigationInfo(requestId, routeID, stringR.getData(), orderType, orderId, xpVehicle.getXpuId(), taskId); log.info("完成下发XPU导航偏离后新的导航数据：" + orderId); } }catch (Exception e){ e.printStackTrace(); }finally { if(null != lock && lock.isHeldByCurrentThread()){ lock.unlock(); } } } /** * 构建导航请求 * * @param orderId * @param orderPhase */ public void buildNaviRequest(String requestId, String orderId, OrderPhase orderPhase, Long sequence, String taskId, String xpuId) { XpVehicle xpVehicle = xpVehicleClient.findByVehicleXpuId(xpuId).getData(); VehiclePosition vehiclePosition = mongoTemplate.findOne(Query.query(Criteria.where("vin").is(xpVehicle.getVehicleVin())), VehiclePosition.class); if(null == vehiclePosition){ log.info("请求导航处理失败，未获取到有效的车辆定位 {}", xpuId); return; } double longitude = vehiclePosition.getPoint().getX(); double latitude = vehiclePosition.getPoint().getY(); //订单类型 1：乘客订单，2=巡游订单 LonLatDto destination = new LonLatDto(); LonLatDto origin = new LonLatDto(); //导航id String routeID = ""; String orderType = ""; if (isTestPlanDemoOrder(orderId)) { // TPD 测试单：参考用户单逻辑下发导航。坐标来源于点位表（operation 服务在 getOrderById 中填充）， // 停车单为两段式（接驾 ROUTE_TO_PICKUP + 送驾 ROUTE_TO_DESTINATION），不停车单仅送驾段。 orderType = OrderType.USER_ORDER; R demoOrderR = xpTestPlanDemoClient.getOrderById(parseTestPlanDemoOrderId(orderId)); XpTestPlanDemoOrder demoOrder = demoOrderR == null ? null : demoOrderR.getData(); if (demoOrder == null) { log.warn("TPD 导航下发跳过，订单或车辆不存在 orderId={}", orderId); return; } //驶往上车点的行程：车辆当前位置 -> 起点 if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) { origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(demoOrder.getStartLon()); destination.setLatitude(demoOrder.getStartLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(parseTestPlanDemoOrderId(orderId)); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(1); dto.setVehicleId(demoOrder.getVehicleId()); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeID = routeVoR.getData().getRouteID(); log.info("TPD 调用地图获取到的行程轨迹(接驾) {} {}", orderId, JSON.toJSONString(routeVoR.getData())); } //驶往下车点的行程：起点 -> 终点 if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) { origin.setLongitude(demoOrder.getStartLon()); origin.setLatitude(demoOrder.getStartLat()); destination.setLongitude(demoOrder.getEndLon()); destination.setLatitude(demoOrder.getEndLat()); OrderRouteAndFareDto dto = new OrderRouteAndFareDto(); dto.setOrderId(parseTestPlanDemoOrderId(orderId)); dto.setOrigin(origin); dto.setDestination(destination); dto.setRouteType(2); dto.setVehicleId(demoOrder.getVehicleId()); R routeVoR = xpMapClient.getOrderDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } routeID = routeVoR.getData().getRouteID(); log.info("TPD 调用地图获取到的行程轨迹(送驾) {} {}", orderId, JSON.toJSONString(routeVoR.getData())); } } else 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(); Object cacheObject = redisService.getCacheObject("CRUISE_ROUTE:" + xpCruiseOrder.getVehicleId()); if(null != cacheObject){ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); log.info("获取到缓存中的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); }else{ 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 routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); log.info("调用地图获取到的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); redisService.setCacheObject("CRUISE_ROUTE:" + xpCruiseOrder.getVehicleId(), JSON.toJSONString(routeVo)); } } } 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 (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) { //直接取缓存中的导航数据推送给车端和乘客 Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_START_DATA:" + testOrder.getVehicleId()); if(null == cacheObject){ origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(testOrder.getStartLon()); destination.setLatitude(testOrder.getStartLat()); RouteAndFareDto dto = new RouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); dto.setVehicleId(testOrder.getVehicleId()); R routeVoR = xpMapClient.getTestDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); log.info("调用地图获取到的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 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(); log.info("获取到缓存中的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); } } //驶往下车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) { //直接取缓存中的导航数据推送给车端和乘客 Object cacheObject = redisService.getCacheObject("TEST_ORDER_NAVIGATION_END_DATA:" + testOrder.getVehicleId()); if(null == cacheObject){ //没有缓存数据，重新规划线路 //查询起点到终点的线路 origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(testOrder.getEndLon()); destination.setLatitude(testOrder.getEndLat()); RouteAndFareDto dto = new RouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); R routeVoR = xpMapClient.getTestDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); log.info("调用地图获取到的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 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(); log.info("获取到缓存中的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); } } }else{ orderType = OrderType.TEST_CRUISE_ORDER; //巡游单 if (orderPhase.equals(OrderPhase.ROUTE_TO_CRUISE)) { /* 构建导航数据给到车端 */ origin.setLongitude(longitude); origin.setLatitude(latitude); destination.setLongitude(testOrder.getEndLon()); destination.setLatitude(testOrder.getEndLat()); RouteAndFareDto dto = new RouteAndFareDto(); dto.setOrderId(Long.parseLong(orderId.substring(4))); dto.setOrigin(origin); dto.setDestination(destination); R routeVoR = xpMapClient.getTestDistanceAndDuration(dto); if (200 != routeVoR.getCode()) { throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); log.info("调用地图获取到的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); } } }else{ orderType = OrderType.USER_ORDER; //正常单 R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); if (200 != xpOrderR.getCode()) { throw new RuntimeException("订单不存在 " + orderId); } XpOrder xpOrder = xpOrderR.getData(); RLock lock = redissonClient.getLock("SET_USER_ORDER_NAVIGATION:" + orderId); try { if(lock.tryLock(10, 10, TimeUnit.SECONDS)){ //驶往上车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_PICKUP)) { if(xpOrder.getStatus() != 2){ return; } Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_START_DATA:" + xpOrder.getId()); if(null == cacheObject){ //查询司机当前位置到起点的线路 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()) { //车辆状态回执未空闲 XpVehicle xpVehicle1 = new XpVehicle(); xpVehicle1.setId(xpVehicle.getId()); xpVehicle1.setState(0); xpVehicleClient.updateVehicleById(xpVehicle1); log.info("请求导航算路失败走改派逻辑，车辆状态修改0 {} {}", xpVehicle.getPlateNumber(), orderId); //业务执行失败，给车端发送取消任务报文，继续执行巡游任务 PushOrderDto pushOrderDto = new PushOrderDto(); pushOrderDto.setEndTaskId(xpOrder.getXpuStartTaskId()); pushOrderDto.setOrderId(orderId); pushOrderDto.setVin(xpVehicle.getVehicleVin()); pushOrderDto.setOperation_type(OrderOperationType.CLOUD_CANCEL_ORDER); orderMissionClient.cancelOrder(pushOrderDto); log.info("请求导航算路失败走改派逻辑，开始执行巡游订单 {}", orderId); //车辆继续巡游 xpOrderClient.addNewCruiseMission(xpVehicle.getVehicleVin(), orderId); log.error("请求导航算路失败走改派逻辑，车辆继续执行巡游 {}", orderId); //将当前车辆添加到禁推列表中 Object cacheObject1 = redisService.getCacheObject("CAN_NOT_RECEIVE_ORDER_VEHICLE:" + xpOrder.getId()); List vinList = new ArrayList<>(); if(Objects.nonNull(cacheObject1)){ vinList = JSON.parseArray(cacheObject1.toString(), String.class); } vinList.add(xpVehicle.getVehicleVin()); redisService.setCacheObject("CAN_NOT_RECEIVE_ORDER_VEHICLE:" + xpOrder.getId(), JSON.toJSONString(vinList), 10L, TimeUnit.MINUTES); //放入队列，重新执行派单 xpOrderClient.againPushOrder(orderId); throw new RuntimeException("获取订单线路和运费失败"); } RouteVo routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); log.info("调用地图获取到的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); //缓存数据，用于后续推给乘客导航线路 redisService.setCacheObject("ORDER_NAVIGATION_START_DATA:" + xpOrder.getId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); }else{ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); log.info("获取到缓存中的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); }; } //驶往下车点的行程 if (orderPhase.equals(OrderPhase.ROUTE_TO_DESTINATION)) { //直接取缓存中的导航数据推送给车端和乘客 Object cacheObject = redisService.getCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId()); if(null == cacheObject){ //没有缓存数据，重新规划线路 //查询起点到终点的线路 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 routeVo = routeVoR.getData(); routeID = routeVo.getRouteID(); log.info("调用地图获取到的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); //缓存导航数据，用于后面的逻辑中给乘客端推送导航数据 redisService.setCacheObject("ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId(), JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS); redisService.setCacheObject("ORDER_NAVIGATION_END_DATA_BACKUP:" + xpOrder.getAppUserId(), JSON.toJSONString(routeVo), 48L, TimeUnit.HOURS); }else{ RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class); routeID = routeVo.getRouteID(); log.info("获取到缓存中的行程轨迹 {} {}", orderId, JSON.toJSONString(routeVo)); } } } }catch (Exception e){ log.info("获取订单导航数据失败 {}", orderId); e.printStackTrace(); }finally { if(null != lock && lock.isHeldByCurrentThread()){ lock.unlock(); } } } if(StringUtils.isNotEmpty(routeID)){ log.info("开始推送导航信息 {}", orderId); 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, xpuId, 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); } private boolean handleTestPlanDemoOrderState(RTCmdType cmd, long sequence, OrderStatusInfo orderStatusInfo) { String orderId = orderStatusInfo.getOrderId(); OrderStatus status = orderStatusInfo.getStatus(); if (!isTestPlanDemoOrder(orderId)) { return false; } // 判断是否为不停靠订单 boolean isNoStopOrder = isTestPlanDemoNoStopOrder(orderId); switch (status) { case ORDER_RECEIVED: // 与用户单一致：车端上报已接收订单，业务侧置为”待出发”(2)。 updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.READY_DEPART); return true; case ORDER_STARTED: // 与用户单一致：开始执行订单，不回写状态，等待后续”前往上车点”等事件推进。 log.info("测试Demo订单开始执行，不回写状态 orderId={}, isNoStop={}", orderId, isNoStopOrder); return true; case ORDER_EN_ROUTE_TO_PICKUP: // 前往上车点，置为”待到达预约地”(3)。 // 注意：“是否停靠”(is_stop_at_dropoff) 只决定【下车点/目的地】是否停车，与接驾段无关； // 不停靠单同样要先驶往上车点接驾，接驾段状态(3/4/5)应正常流转，不得因 isNoStop 跳过 // （plan337 实测：不停靠单上报 EN_ROUTE_TO_PICKUP 却被忽略，状态从待出发(2)直接跳到服务中(6)）。 updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.PICK_UP); return true; case ORDER_WAITING_FOR_PICKUP: updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.PICK_UP_GUESTS); return true; case ORDER_PASSENGER_ONBOARD: updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.PASSENGER_BOARDED); return true; case ORDER_EN_ROUTE_TO_DESTINATION: // 前往目的地：从状态5（乘客已上车）推进到状态6（服务中）。 // 不停靠单同样会经过接驾段(3/4/5)，此处不再按 isNoStop 区分。 updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.IN_SERVICE); return true; case ORDER_WAITING_FOR_DROPOFF: updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.SEE_OFF_GUESTS); return true; case ORDER_ARRIVED_AT_DESTINATION: // 与用户单一致：到达目的地仅记录日志，不在此处置完成。 // 停车单的完成由 ORDER_COMPLETED 推进，不停车单的完成由“即将到达终点”事件推进。 log.info("测试Demo订单到达目的地，等待完成事件 orderId={}", orderId); return true; case ORDER_COMPLETED: updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.COMPLETED); return true; case ORDER_CANCELLED: JSONObject reassignCancelContext = resolveReassignCancelContext(orderId, orderStatusInfo); if (Objects.nonNull(reassignCancelContext)) { log.info("测试Demo改派旧车取消回调，跳过订单取消处理，orderId: {}, taskId: {}, cancelScene: {}, context: {}", orderId, orderStatusInfo.getTaskId(), reassignCancelContext.getString("cancelScene"), reassignCancelContext); String responseVin = resolveReassignCancelResponseVin(orderId, reassignCancelContext); if (StringUtils.isNotEmpty(responseVin)) { responseMessage(cmd, sequence, responseVin); } else { log.warn("测试Demo改派旧车取消回调未找到响应VIN，orderId: {}, taskId: {}, context: {}", orderId, orderStatusInfo.getTaskId(), reassignCancelContext); } return true; } if (isTerminalTestPlanDemoOrder(orderId)) { log.info("测试Demo订单已终态，忽略车机取消状态回调 orderId={}", orderId); return true; } updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.CANCELED); return true; default: log.info("测试Demo订单状态不处理 orderId={}, status={}", orderId, status); return true; } } private void updateTestPlanDemoOrderStatus(String orderId, OrderStatusEnum status) { Long demoOrderId = parseTestPlanDemoOrderId(orderId); R result = xpTestPlanDemoClient.updateOrderStatus(demoOrderId, status.getCode()); if (result == null || R.isError(result)) { log.warn("更新测试Demo订单状态失败 orderId={}, status={}, msg={}", orderId, status.getCode(), result == null ? null : result.getMsg()); } } private void responseTestPlanDemoOrder(RTCmdType cmd, long sequence, String orderId) { Long demoOrderId = parseTestPlanDemoOrderId(orderId); R orderResult = xpTestPlanDemoClient.getOrderById(demoOrderId); XpTestPlanDemoOrder order = orderResult == null ? null : orderResult.getData(); if (order == null || order.getVehicleId() == null) { log.warn("测试Demo订单到达目的地响应失败，订单或车辆不存在 orderId={}", orderId); return; } R vehicleResult = xpVehicleClient.queryById(order.getVehicleId()); XpVehicle vehicle = vehicleResult == null ? null : vehicleResult.getData(); if (vehicle == null) { log.warn("测试Demo订单到达目的地响应失败，车辆不存在 orderId={}, vehicleId={}", orderId, order.getVehicleId()); return; } String vin = vehicle.getVehicleVin(); responseMessage(cmd, sequence, vin); } /** * 订单是否已进入终态（已完成或已取消）。 * 用于车端晚到事件（ORDER_CANCELLED、即将到达终点）的前置拦截：订单已终态时直接忽略， * 避免对已取消订单仍走"提前置完成"或重复取消处理（plan253 双派排查衍生加固，原仅判 COMPLETED 漏 CANCELED）。 */ private boolean isTerminalTestPlanDemoOrder(String orderId) { Long demoOrderId = parseTestPlanDemoOrderId(orderId); R orderResult = xpTestPlanDemoClient.getOrderById(demoOrderId); XpTestPlanDemoOrder order = orderResult == null ? null : orderResult.getData(); if (order == null) { return false; } return Integer.valueOf(OrderStatusEnum.COMPLETED.getCode()).equals(order.getOrderStatus()) || Integer.valueOf(OrderStatusEnum.CANCELED.getCode()).equals(order.getOrderStatus()); } private boolean isTestPlanDemoOrder(String orderId) { return StringUtils.isNotEmpty(orderId) && orderId.startsWith(TEST_PLAN_DEMO_ORDER_PREFIX); } private Long parseTestPlanDemoOrderId(String orderId) { return Long.parseLong(orderId.substring(TEST_PLAN_DEMO_ORDER_PREFIX.length())); } /** * 判断是否为不停靠的测试Demo订单。 * 不停靠订单没有接驾阶段，直接从"待出发"进入"服务中"。 */ private boolean isTestPlanDemoNoStopOrder(String orderId) { if (!isTestPlanDemoOrder(orderId)) { return false; } try { Long demoOrderId = parseTestPlanDemoOrderId(orderId); R stopResult = xpTestPlanDemoClient.isStopAtDropoff(demoOrderId); if (stopResult != null && stopResult.getData() != null) { // isStopAtDropoff 返回 false 表示不停靠 return !stopResult.getData(); } } catch (Exception e) { log.error("判断测试Demo订单是否停靠失败 orderId={}", orderId, e); } // 默认认为是停靠订单（保守处理） return false; } /** * 订单状态更新事件 * @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(); if (handleTestPlanDemoOrderState(cmd, sequence, orderStatusInfo)) { return; } clearPickupExceptionLiveLocationOnStatus(orderId, status); 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()); } } 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); R updateResult = xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.PICK_UP.getCode()); if (updateResult != null && updateResult.getCode() == Constants.SUCCESS) { reportOperateDepartToMinistry(orderId); } } } 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()); } } break; case ORDER_PASSENGER_ONBOARD: //乘客已上车 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()); } } 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()); } } break; case ORDER_ARRIVED_AT_DESTINATION: //到达目的地 log.info("到达目的地：{}", orderId); if(orderId.contains("TEST")){ //测试单到达目的地：仅响应车端 ACK，状态更新与链式回调已迁移到 ORDER_NEAR_DESTINATION_EVENT XpTestOrder testOrder = testOrderClient.queryOrderById(Long.parseLong(orderId.substring(4))).getData(); if(Objects.nonNull(testOrder)){ XpVehicle xpVehicle = xpVehicleClient.queryById(testOrder.getVehicleId()).getData(); if(Objects.nonNull(xpVehicle)){ //响应消息 String vin = xpVehicle.getVehicleVin(); responseMessage(cmd, sequence, vin); } } } else if (!orderId.contains("XU")) { reportOperateArriveToMinistry(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()); } } break; case ORDER_DROPOFF_COMPLETED: //下车完成，不需要处理状态，由后续的乘客主动触发或者定时任务触发 log.info("乘客已下车，不做任何处理：{}", orderId); break; case ORDER_COMPLETED: //已完成 log.info("已完成，不做任何处理：{}", orderId); if(!orderId.contains("XU") && !orderId.contains("TEST")){ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.COMPLETED.getCode()); 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); } } }else if(orderId.contains("XU")){ buildPushOrderStatus(cmd, orderId, OrderStatusEnum.COMPLETED, sequence); } break; case ORDER_CANCELLED: //已取消 if(!orderId.contains("XU")){ if(!orderId.contains("TEST")){ //发送巡游订单 XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData(); if(null != xpOrder && xpOrder.getStatus() > 7){ xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.COMPLETED.getCode()); } } 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(orderId, reassignCancelContext); if (StringUtils.isNotEmpty(responseVin)) { responseMessage(cmd, sequence, responseVin); } else { log.warn("改派旧车取消回调未找到响应VIN，orderId: {}, taskId: {}, context: {}", orderId, taskId, reassignCancelContext); } return; } } break; case ORDER_ARRIVED_AT_CRUISE_DESTINATION: log.info("巡游订单到达终点：{}", orderId); break; case ORDER_PAUSED: //订单暂停 log.info("订单暂停，不做任何处理：{}", orderId); break; default: break; } // 车端订单状态变更埋点 if (!orderId.contains("TEST")) { try { String eventVin = null; String orderType = "user_order"; if (orderId.contains("XU")) { orderType = "cruise_order"; String cruiseOrderId = orderId.substring(2); XpCruiseOrder cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(cruiseOrderId)).getData(); if (Objects.nonNull(cruiseOrder) && Objects.nonNull(cruiseOrder.getVehicleId())) { XpVehicle eventVehicle = xpVehicleClient.queryById(cruiseOrder.getVehicleId()).getData(); if (Objects.nonNull(eventVehicle)) { eventVin = eventVehicle.getVehicleVin(); } } } else { XpOrder eventOrder = xpOrderClient.queryOrderById(Long.valueOf(orderId)).getData(); if (Objects.nonNull(eventOrder) && Objects.nonNull(eventOrder.getVehicleId())) { XpVehicle eventVehicle = xpVehicleClient.queryById(eventOrder.getVehicleId()).getData(); if (Objects.nonNull(eventVehicle)) { eventVin = eventVehicle.getVehicleVin(); } } } if (StringUtils.isNotEmpty(eventVin)) { VehicleEventDTO vehicleEventDTO = new VehicleEventDTO(); vehicleEventDTO.setVin(eventVin); vehicleEventDTO.setXpuId(""); vehicleEventDTO.setOrderId(orderId); vehicleEventDTO.setEventKey("RT_web_vehicle_status_changed"); vehicleEventDTO.setEventTime(System.currentTimeMillis()); Map attributes = new HashMap<>(); attributes.put("vin", eventVin); attributes.put("order_id", orderId); attributes.put("order_type", orderType); attributes.put("timestamp", vehicleEventDTO.getEventTime()); attributes.put("main_status", "ONLINE"); attributes.put("sub_status", status.name()); attributes.put("prev_sub_status", ""); VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(eventVin)), VehiclePosition.class); if (Objects.nonNull(vehiclePosition)) { attributes.put("odometer_km", vehiclePosition.getTotalOdometer()); } else { attributes.put("odometer_km", ""); } vehicleEventDTO.setAttributes(attributes); vehicleBuriedPointClient.sendVehicleEvent(vehicleEventDTO); } } catch (Exception e) { log.error("车端订单状态变更埋点异常，orderId={}, status={}", orderId, status, e); } } } } // /** // * 用户下车信息事件 // * // * @param msgContent // */ // private void dropoffInfoEvent(MsgContent msgContent) { // DropoffInfo dropoffInfo = msgContent.getDropoffInfo(); // if (isTestPlanDemoOrder(dropoffInfo.getOrderId())) { // // TPD 测试单：下车完成不需要处理（续单由“即将到达终点”或完成事件触发）。 // log.info("测试Demo订单下车信息不处理：{} {}", dropoffInfo.getOrderId(), dropoffInfo.getDropoffStatus()); // return; // } // if (!DropoffStatus.DROPOFF_COMPLETED.equals(dropoffInfo.getDropoffStatus())) { // log.info("用户下车信息未完成，不触发巡游续单：{} {}", dropoffInfo.getOrderId(), dropoffInfo.getDropoffStatus()); // return; // } // dispatchCruiseAfterDropoff(dropoffInfo.getOrderId()); // } private void reportOperateDepartToMinistry(String orderId) { try { if (StringUtils.isEmpty(orderId) || orderId.contains("XU") || orderId.contains("TEST")) { return; } R result = xpOrderClient.reportOperateDepartToMinistry(Long.parseLong(orderId)); if (Objects.isNull(result) || result.getCode() != Constants.SUCCESS) { log.warn("提交监管平台YYCF上报失败，orderId={}, msg={}", orderId, Objects.nonNull(result) ? result.getMsg() : null); } } catch (Exception e) { log.error("提交监管平台YYCF上报异常，orderId={}", orderId, e); } } private void reportOperateArriveToMinistry(String orderId) { try { if (StringUtils.isEmpty(orderId) || orderId.contains("XU") || orderId.contains("TEST")) { return; } R result = xpOrderClient.reportOperateArriveToMinistry(Long.parseLong(orderId)); if (Objects.isNull(result) || result.getCode() != Constants.SUCCESS) { log.warn("提交监管平台YYDD上报失败，orderId={}, msg={}", orderId, Objects.nonNull(result) ? result.getMsg() : null); } } catch (Exception e) { log.error("提交监管平台YYDD上报异常，orderId={}", orderId, e); } } // // /** // * 收到车端下车完成信号后，再推进订单完成和续巡游，避免车辆仍在等待乘客下车时被新巡游任务带走。 // * // * @param orderId // */ // private void dispatchCruiseAfterDropoff(String orderId) { // if (StringUtils.isEmpty(orderId) || orderId.contains("XU") || orderId.contains("TEST") // || isTestPlanDemoOrder(orderId)) { // // TPD 测试单不参与下车完成续巡游逻辑：下车完成不处理，避免 Long.parseLong("TPDxxx") 抛 NumberFormatException。 // return; // } // String confirmedKey = ORDER_DROPOFF_CONFIRMED_PREFIX + orderId; // String dispatchedKey = ORDER_DROPOFF_CRUISE_DISPATCHED_PREFIX + orderId; // redisService.setCacheObject(confirmedKey, "1", 24L, TimeUnit.HOURS); // Boolean dispatched = redisService.hasKey(dispatchedKey); // if (Boolean.TRUE.equals(dispatched)) { // log.info("下车完成后的巡游续单已处理，跳过重复事件：{}", orderId); // return; // } // redisService.setCacheObject(dispatchedKey, "1", 24L, TimeUnit.HOURS); // R updateResult = xpOrderClient.updateOrderStatus(Long.parseLong(orderId), OrderStatusEnum.COMPLETED.getCode()); // if (Objects.isNull(updateResult) || updateResult.getCode() != Constants.SUCCESS) { // redisService.deleteObject(dispatchedKey); // log.warn("下车完成后推进订单完成失败，暂不续巡游：{} {}", orderId, Objects.nonNull(updateResult) ? updateResult.getMsg() : null); // } // } public void xuOrder(String orderId, Long sequence, RTCmdType cmd){ //到达巡游终点 if(orderId.contains("XU")){ endXpCruiseOrder(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(); //响应消息 String vin = xpVehicle.getVehicleVin(); responseMessage(cmd, sequence, vin); //测试巡游单处理逻辑 testOrderClient.updateOrderStatus(Long.parseLong(orderId.substring(4)), OrderStatusEnum.COMPLETED.getCode()); redisService.setCacheObject("TEST_ORDER_STATUS:" + orderId.substring(4), OrderStatusEnum.COMPLETED.getCode(), 24L, TimeUnit.HOURS); // 链式回调：如果是批量测试订单，触发下一段订单创建 if (testOrder.getTestPlanId() != null) { try { xpTestPlanClient.chainCallback(Long.parseLong(orderId.substring(4)), testOrder.getTestPlanId().longValue()); } catch (Exception e) { log.error("链式回调失败, orderId={}, testPlanId={}", orderId.substring(4), testOrder.getTestPlanId(), e); } } } } } private void endXpCruiseOrder(String orderId){ String cruiseOrderId = orderId.substring(2); XpCruiseOrder xpCruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(cruiseOrderId)).getData(); Long vehicleId = xpCruiseOrder.getVehicleId(); R xpVehicleR = xpVehicleClient.queryById(vehicleId); XpVehicle xpVehicle = xpVehicleR.getData(); //响应消息 // 运营中 继续执行 if(xpVehicle.getOperatingStatus()==1 && xpVehicle.getState()==0 && xpVehicle.getAcceptOrTestStatus() == 1){ 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(0); newXpCruiseOrder.setLastOrderId(xpCruiseOrder.getId()); newXpCruiseOrder.setXpuTaskId(snowflake.nextId() + ""); newXpCruiseOrder.setOrderReverse(xpCruiseOrder.getOrderReverse()); // 通过车找终点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 && 200 == cruisePoint.getCode() && cruisePoint.getData() != null){ XpCruiseOrder cruiseOrder1 = cruisePoint.getData(); //不停车替换新任务 cruiseOrder1.setLastOrderId1("XU" + xpCruiseOrder.getId()); cruiseOrder1.setLastXpuTaskId(xpCruiseOrder.getXpuTaskId()); cruiseOrder1.setTaskActionType(TaskActionType.kTerminateForNewTask); xpCruiseOrderClient.addCruiseOrder(cruiseOrder1); //完成当前巡游订单 xpCruiseOrderClient.completeCruiseOrder(xpCruiseOrder.getId()); } else { // 有效点位不足或无可用路线时，operation 会返回空数据表示终止本次续单。 log.info("巡游计划未返回下一笔巡游单，停止续单，orderId={}, vehicleId={}", orderId, xpCruiseOrder.getVehicleId()); } } } /** * 构建订单状态推送消息 * @param orderId * @param status */ public void buildPushOrderStatus(RTCmdType cmd, String orderId, OrderStatusEnum status, Long sequence){ if(orderId.contains("XU")){ //巡游单 //到达目的地 if(OrderStatusEnum.COMPLETED.equals(status)){ //推送新的巡游单，再完成旧的巡游单 endXpCruiseOrder(orderId); R cruiseOrder = xpCruiseOrderClient.getCruiseOrder(Long.parseLong(orderId.substring(2))); if(cruiseOrder!=null){ Long vehicleId = cruiseOrder.getData().getVehicleId(); R xpVehicleR = xpVehicleClient.queryById(vehicleId); XpVehicle xpVehicle = xpVehicleR.getData(); //响应消息 String vin = xpVehicle.getVehicleVin(); responseMessage(cmd, sequence, vin); // 封装埋点消息实体 VehicleEventDTO vehicleEventDTO = new VehicleEventDTO(); vehicleEventDTO.setVin(vin); vehicleEventDTO.setXpuId(""); vehicleEventDTO.setOrderId(orderId.substring(2)); 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{ //普通单 clearPickupExceptionLiveLocationOnStatus(orderId, status); R xpOrderR = xpOrderClient.queryOrderById(Long.parseLong(orderId)); XpOrder xpOrder = xpOrderR.getData(); if(xpOrder.getStatus() == 0){ return; } XpVehicle xpVehicle = xpVehicleClient.queryById(xpOrder.getVehicleId()).getData(); //响应消息 String vin = xpVehicle.getVehicleVin(); responseMessage(cmd, sequence, vin); //构建推送消息 Map params = new HashMap<>(); Map content = new HashMap<>(); content.put("orderId", orderId.toString()); content.put("status", status.getCode()); content.put("hasPreviousOrder", redisService.hasPreviousOrder(orderId.toString())); 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 buildPickupExceptionForStatusPush(XpOrder order, String vin, PullOverInfo pullOverInfo) { JSONObject pickupException = new JSONObject(); pickupException.put("enabled", false); if (order == null || StringUtils.isEmpty(vin) || order.getStartLon() == null || order.getStartLat() == null || pullOverInfo == null) { return pickupException; } try { VehiclePosition vehiclePosition = mongoTemplate.findOne(new Query(Criteria.where("vin").is(vin)), VehiclePosition.class); if (vehiclePosition == null || vehiclePosition.getPoint() == null) { return pickupException; } double vehicleLon = vehiclePosition.getPoint().getX(); double vehicleLat = vehiclePosition.getPoint().getY(); pickupException.put("enabled", true); pickupException.put("type", PICKUP_EXCEPTION_TYPE_REMOTE_STOP); pickupException.put("vehicleLon", vehicleLon); pickupException.put("vehicleLat", vehicleLat); pickupException.put("pickupLon", order.getStartLon()); pickupException.put("pickupLat", order.getStartLat()); pickupException.put("offsetDistance", Math.round(pullOverInfo.getDistanceToEndPointM())); pickupException.put("message", PICKUP_EXCEPTION_MESSAGE); R refreshResult = xpOrderClient.refreshEndNavigationRoute(order.getId(), vehicleLon, vehicleLat, vin); if (refreshResult == null || refreshResult.getCode() != Constants.SUCCESS) { log.warn("靠边停车异常刷新送驾路线失败 orderId={}, msg={}", order.getId(), refreshResult == null ? null : refreshResult.getMsg()); } } catch (Exception e) { log.warn("构建状态推送接驾异常信息失败 orderId={}", order.getId(), e); } return pickupException; } /** * 当前 cancelOrder 指令是按 taskId 记录取消上下文。 * 改派旧车取消回调优先读取该上下文，避免依赖订单当前车/状态造成竞态误判。 */ 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; } String taskId = orderStatusInfo.getTaskId(); // 先按 taskId 读取当前生效的取消上下文，再兼容历史按 orderId 存储的改派上下文。 JSONObject cancelContext = getCancelOrderContext(taskId); if (cancelContext == null) { cancelContext = getReassignCancelContext(orderId); } if (cancelContext == null) { log.info("改派取消回调未命中专用上下文，orderId: {}, taskId: {}, missReason: context_not_found", orderId, taskId); return null; } String cancelScene = cancelContext.getString("cancelScene"); if (CancelOrderSceneEnum.REASSIGN_OLD_VEHICLE != CancelOrderSceneEnum.fromCode(cancelScene)) { log.info("改派取消回调未命中专用上下文，orderId: {}, taskId: {}, missReason: scene_not_match, cancelScene: {}, context: {}", orderId, taskId, cancelScene, cancelContext); return null; } String contextOrderId = cancelContext.getString("orderId"); if (StringUtils.isNotEmpty(contextOrderId) && !Objects.equals(contextOrderId, orderId)) { log.info("改派取消回调未命中专用上下文，orderId: {}, taskId: {}, missReason: order_not_match, contextOrderId: {}, context: {}", orderId, taskId, contextOrderId, cancelContext); return null; } return cancelContext; } /** * 当前 cancelOrder 指令是按 taskId 记录取消上下文。 * 改派旧车取消回调优先读取该上下文，避免依赖订单当前车/状态造成竞态误判。 */ private JSONObject getCancelOrderContext(String taskId) { if (StringUtils.isEmpty(taskId)) { return null; } Object cacheObject = redisService.getCacheObject(CANCEL_ORDER_CONTEXT_PREFIX + taskId); if (cacheObject == null) { return null; } return JSON.parseObject(cacheObject.toString()); } private String resolveReassignCancelResponseVin(String orderId, JSONObject cancelContext) { if (cancelContext == null) { return null; } String vin = cancelContext.getString("vin"); if (StringUtils.isNotEmpty(vin)) { return vin; } 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 newVehicleId.toString(); } 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(); log.info("改派取消回调未命中专用上下文，orderId: {}, currentVehicleId: {}, newVehicleId: {}, currentStatus: {}", orderId, xpOrder.getVehicleId(), newVehicleId, currentStatus); return xpVehicle == null ? null : xpVehicle.getVehicleVin(); } /** * 云端下发导航到XPU * @param requestId XPU请求id * @param routeId 高德导航id * @param routeDataContent 高德导航原数据 * @param orderId 订单id * @param xpuId XPU设备id */ public void cloudPushNavigationInfo(String requestId, String routeId, String routeDataContent, String orderType, String orderId, 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)); } //靠边停车处理逻辑 public void pulloverStatusEvent(MsgContent msgContent){ //开始处理消息对应的业务逻辑 //XPU→云端的靠边停车状态事件(cmd=16002)业务数据挂在 pull_over_info_event(字段230)， //而非 CDU 侧的 pullover_info(字段402)；二者属于同一 oneof，缺失该事件直接报错结束。 if (!msgContent.hasPullOverInfoEvent()) { log.error("靠边停车状态事件缺少 pull_over_info_event，无法处理，msgContent={}", JSON.toJSONString(msgContent)); return; } PullOverInfo pulloverInfo = msgContent.getPullOverInfoEvent().getPullOverInfo(); String orderId = pulloverInfo.getOrderId(); String taskId = pulloverInfo.getTaskId(); PullOverStatus pullOverStatus = pulloverInfo.getPullOverStatus(); PullOverReasons pulloverReasons = pulloverInfo.getPulloverReasons(); double distanceToEndPointM = pulloverInfo.getDistanceToEndPointM(); OrderPhase orderPhase = pulloverInfo.getOrderPhase(); log.info("订单ID：{}", orderId); log.info("任务ID：{}", taskId); //获取系统异常订单超时配置，创建定时任务取消订单 if (isTestPlanDemoOrder(orderId)) { return; } if(!orderId.contains("XU")){ //靠边停车完成后，创建定时任务处理后续业务逻辑 XpSystemSetting xpSystemSetting = systemSettingClient.getSystemSettingOne().getData(); Map jobMaps = new ImmutableMap.Builder() .put("id", orderId) .put("orderType", orderId.contains("TEST") ? OrderType.TEST_ORDER : OrderType.USER_ORDER) .build(); if (isPullOverCompleted(pullOverStatus, pulloverReasons, distanceToEndPointM, orderPhase)) { Boolean firstHandle = redisService.setIfAbsent(PICKUP_EXCEPTION_HANDLED_PREFIX + orderId, "1", 24L, TimeUnit.HOURS); // 车端可能重复上报异常靠边停车完成，同一订单只处理首次，避免重复推送和刷新超时任务。 if (!Boolean.TRUE.equals(firstHandle)) { log.info("异常靠边停车已处理，跳过重复事件 orderId={}, taskId={}", orderId, taskId); return; } // 接驾阶段远距离靠边停车后，用户在异常处理窗口内取消不计入取消风控责任。 redisService.setCacheObject(PICKUP_EXCEPTION_FREE_CANCEL_PREFIX + orderId, "1", (long)PICKUP_EXCEPTION_FREE_CANCEL_EXTEND_MINUTES, TimeUnit.MINUTES); extendPickupCancelJob(orderId); markPickupExceptionLiveLocation(orderId); pushPickupExceptionForPulloverStatus(orderId, pulloverInfo); QuartzManager.addJob( OrderExceptionJob.class, (OrderExceptionJob.name + orderId).toUpperCase(), TimeJobType.ERROR_ORDER_CANCEL, DateUtils.localDateTimeToDate(LocalDateTime.now().plusMinutes(xpSystemSetting.getOrderExceptionalTimeout()+PICKUP_EXCEPTION_FREE_CANCEL_EXTEND_MINUTES)), jobMaps ); } else if (pullOverStatus.getNumber() == PullOverStatus.PULL_OVER_COMPLETED.getNumber()) { String orderExceptionJobName = (OrderExceptionJob.name + orderId).toUpperCase(); // 已存在同名异常订单任务时不覆盖，避免普通靠边停车事件缩短接驾异常延长后的超时时间。 if (QuartzManager.existsJob(orderExceptionJobName, TimeJobType.ERROR_ORDER_CANCEL)) { log.info("靠边停车普通超时任务已存在，跳过重复添加 orderId={}", orderId); return; } QuartzManager.addJob( OrderExceptionJob.class, orderExceptionJobName, TimeJobType.ERROR_ORDER_CANCEL, DateUtils.localDateTimeToDate(LocalDateTime.now().plusMinutes(xpSystemSetting.getOrderExceptionalTimeout())), jobMaps ); } } } private void extendPickupCancelJob(String orderId) { try { R result = xpOrderClient.extendPickupCancelJob(Long.parseLong(orderId), PICKUP_EXCEPTION_FREE_CANCEL_EXTEND_MINUTES); if (result == null || result.getCode() != Constants.SUCCESS) { log.warn("接驾异常延长待客上车取消任务失败 orderId={}, msg={}", orderId, result == null ? null : result.getMsg()); } } catch (Exception e) { log.warn("接驾异常延长待客上车取消任务异常 orderId={}", orderId, e); } } private boolean isPullOverCompleted(PullOverStatus pullOverStatus, PullOverReasons pulloverReasons, double distanceToEndPointM, OrderPhase orderPhase) { return pullOverStatus == PullOverStatus.PULLOVER_COMPLETE_WITH_PARKING && (distanceToEndPointM > PICKUP_EXCEPTION_MIN_DISTANCE_METERS && distanceToEndPointM <= PICKUP_EXCEPTION_MAX_DISTANCE_METERS) && orderPhase == OrderPhase.ROUTE_TO_PICKUP; } private void markPickupExceptionLiveLocation(String orderId) { try { redisService.setCacheObject(PICKUP_EXCEPTION_LIVE_LOCATION_PREFIX + orderId, "1", (long) PICKUP_EXCEPTION_LIVE_LOCATION_MINUTES, TimeUnit.MINUTES); } catch (Exception e) { log.warn("接驾异常实时定位标记写入失败 orderId={}", orderId, e); } } private void pushPickupExceptionLiveLocation(String orderId, double longitude, double latitude) { try { if (StringUtils.isEmpty(orderId) || !Boolean.TRUE.equals(redisService.hasKey(PICKUP_EXCEPTION_LIVE_LOCATION_PREFIX + orderId))) { return; } R orderResult = xpOrderClient.queryOrderById(Long.parseLong(orderId)); XpOrder order = orderResult == null ? null : orderResult.getData(); if (order == null || order.getAppUserId() == null) { clearPickupExceptionLiveLocation(orderId); return; } JSONObject content = new JSONObject(); content.put("orderId", orderId); content.put("lon", longitude); content.put("lat", latitude); Map params = new HashMap<>(); params.put("type", PushMessageTypeEnum.VEHICLE_LOCATION.getDesc()); params.put("content", content.toJSONString()); params.put("timestamp", System.currentTimeMillis() + ""); SendMessageDTO dto = new SendMessageDTO(); dto.setAppUserId(order.getAppUserId() + ""); dto.setMessage(JSONObject.toJSONString(params)); R result = xpPushMessageClient.sendVehicleLocation(dto); if (result == null || result.getCode() != Constants.SUCCESS) { log.info("接驾异常实时定位推送失败 orderId={}, msg={}", orderId, result == null ? null : result.getMsg()); } } catch (Exception e) { log.warn("接驾异常实时定位推送异常 orderId={}", orderId, e); } } private void clearPickupExceptionLiveLocationOnStatus(String orderId, OrderStatus status) { if (StringUtils.isEmpty(orderId) || orderId.contains("XU") || orderId.contains("TEST") || isTestPlanDemoOrder(orderId)) { return; } if (status == OrderStatus.ORDER_PASSENGER_ONBOARD || status == OrderStatus.ORDER_EN_ROUTE_TO_DESTINATION || status == OrderStatus.ORDER_CANCELLED || status == OrderStatus.ORDER_COMPLETED) { clearPickupExceptionLiveLocation(orderId); } } private void clearPickupExceptionLiveLocationOnStatus(String orderId, OrderStatusEnum status) { if (StringUtils.isEmpty(orderId) || orderId.contains("XU") || orderId.contains("TEST") || isTestPlanDemoOrder(orderId)) { return; } if (OrderStatusEnum.PASSENGER_BOARDED.equals(status) || OrderStatusEnum.IN_SERVICE.equals(status) || OrderStatusEnum.CANCELED.equals(status) || OrderStatusEnum.COMPLETED.equals(status)) { clearPickupExceptionLiveLocation(orderId); } } private void clearPickupExceptionLiveLocation(String orderId) { try { if (StringUtils.isNotEmpty(orderId) && Boolean.TRUE.equals(redisService.hasKey(PICKUP_EXCEPTION_LIVE_LOCATION_PREFIX + orderId))) { redisService.deleteObject(PICKUP_EXCEPTION_LIVE_LOCATION_PREFIX + orderId); } } catch (Exception e) { log.warn("接驾异常实时定位标记清理失败 orderId={}", orderId, e); } } private void pushPickupExceptionForPulloverStatus(String orderId, PullOverInfo pulloverInfo) { try { R orderResult = xpOrderClient.queryOrderById(Long.parseLong(orderId)); XpOrder order = orderResult == null ? null : orderResult.getData(); if (order == null || order.getStatus() == null || order.getStatus() == 0 || order.getAppUserId() == null) { log.info("靠边停车完成接驾异常推送跳过，订单不存在或状态无效 orderId={}", orderId); return; } R vehicleResult = xpVehicleClient.queryById(order.getVehicleId()); XpVehicle vehicle = vehicleResult == null ? null : vehicleResult.getData(); if (vehicle == null || StringUtils.isEmpty(vehicle.getVehicleVin())) { log.info("靠边停车完成接驾异常推送跳过，车辆不存在或VIN为空 orderId={}", orderId); return; } Map content = new HashMap<>(); content.put("orderId", orderId); content.put("status", order.getStatus()); content.put("hasPreviousOrder", redisService.hasPreviousOrder(orderId)); content.put("pickupException", buildPickupExceptionForStatusPush(order, vehicle.getVehicleVin(), pulloverInfo)); Map params = new HashMap<>(); params.put("type", PushMessageTypeEnum.ORDER_STATUS.getDesc()); params.put("content", JSONObject.toJSONString(content)); params.put("timestamp", System.currentTimeMillis() + ""); SendMessageDTO dto = new SendMessageDTO(); dto.setAppUserId(order.getAppUserId() + ""); dto.setMessage(JSONObject.toJSONString(params)); R result = xpPushMessageClient.sendStatus(dto); if (result == null || result.getCode() != Constants.SUCCESS) { log.info("靠边停车完成接驾异常推送失败 orderId={}, msg={}", orderId, result == null ? null : result.getMsg()); } } catch (Exception e) { log.warn("靠边停车完成接驾异常推送异常 orderId={}", orderId, e); } } /** * 车辆即将到达终点 * @param msgContent */ public void orderNearDestinationEvent(RTCmdType cmd, MsgContent msgContent){ //开始处理消息对应的业务逻辑 OrderNearDestinationEvent orderNearDestinationEvent = msgContent.getOrderNearDestinationEvent(); OrderNearDestination orderNearDestination = orderNearDestinationEvent.getOrderNearDestination(); com.xiaopeng.api.vo.automaticPilot.protobuf.OrderType orderType = orderNearDestination.getOrderType(); long sequence = msgContent.getHeader().getSequence(); String orderId = orderNearDestination.getOrderId(); if (isTestPlanDemoOrder(orderId)) { handleTestPlanDemoNearDestination(cmd, sequence, orderId); return; } //巡游任务，提前下发新的巡游 if(orderType == com.xiaopeng.api.vo.automaticPilot.protobuf.OrderType.ORDER_TYPE_CRUISE){ xuOrder(orderId, sequence, cmd); } } private void handleTestPlanDemoNearDestination(RTCmdType cmd, long sequence, String orderId) { if (isTerminalTestPlanDemoOrder(orderId)) { log.info("测试Demo订单已终态，忽略即将到达终点事件 orderId={}", orderId); return; } // 先响应车端 ACK responseTestPlanDemoOrder(cmd, sequence, orderId); // 是否停车由所属测试计划决定（operation 服务按 planId 解析 isStopAtDropoff），查询失败时默认按停车处理 R stopResult = xpTestPlanDemoClient.isStopAtDropoff(parseTestPlanDemoOrderId(orderId)); boolean stopAtDropoff = stopResult == null || R.isError(stopResult) || !Boolean.FALSE.equals(stopResult.getData()); if (stopAtDropoff) { // 停车单：与用户单一致，即将到达终点不做额外处理，完成由 ORDER_COMPLETED 事件推进。 log.info("测试Demo停车单即将到达终点，等待完成事件推进 orderId={}", orderId); return; } // 不停车单：车辆将直接驶过终点（DRIVE_THROUGH），参考测试单链式逻辑—— // 提前置完成，由 operation 服务在完成时为该车派发下一单。 updateTestPlanDemoOrderStatus(orderId, OrderStatusEnum.COMPLETED); log.info("测试Demo不停车单即将到达终点，提前置完成并触发下一单 orderId={}", orderId); } public void userOperationInfoEvent(RTCmdType cmd, MsgContent msgContent){ //开始处理消息对应的业务逻辑 UserOperationEvent userOperationEvent = msgContent.getUserOperationEvent(); RtEventType eventType = userOperationEvent.getEventType(); //开门关门 if(eventType == RtEventType.kEventUserOperation){ UserOperationInfo userOperationInfo = userOperationEvent.getUserOperationInfo(); String orderId = userOperationInfo.getOrderId(); OrderStatus orderStatus = userOperationInfo.getOrderStatus(); int operationType = userOperationInfo.getOperationType(); if(1 != operationType || StringUtils.isEmpty(orderId)){ return; } if(!orderId.contains("XU")){ if(orderId.contains("TEST")){ }else{ switch (orderStatus){ case ORDER_STATUS_UNSPECIFIED: break; case ORDER_RECEIVED: break; case ORDER_STARTED: break; case ORDER_EN_ROUTE_TO_PICKUP: break; case ORDER_ARRIVED_AT_PICKUP: break; case ORDER_WAITING_FOR_PICKUP: XpOrder xpOrder = xpOrderClient.queryOrderById(Long.parseLong(orderId)).getData(); if(null == xpOrder){ return; } log.info("等待乘客上车 {}", orderId); xpOrderClient.removeOrderJob(Long.parseLong(orderId), OrderStatusEnum.PICK_UP_GUESTS.getCode()); xpOrderClient.removeOrderJob(Long.parseLong(orderId), OrderStatusEnum.PASSENGER_BOARDED.getCode()); //发送socket消息 JSONObject jsonObject = new JSONObject(); jsonObject.put("type", "USER_OPEN_DOOR"); jsonObject.put("timestamp", System.currentTimeMillis()); JSONObject content = new JSONObject(); jsonObject.put("content", content.toJSONString()); SendMessageDTO messageDTO = new SendMessageDTO(); messageDTO.setAppUserId(xpOrder.getAppUserId().toString()); messageDTO.setMessage(jsonObject.toJSONString()); xpPushMessageClient.userOpenDoor(messageDTO); redisService.setCacheObject("USER_OPEN_DOOR:" + orderId, "1", 24L, TimeUnit.HOURS); break; case ORDER_PASSENGER_ONBOARD: log.info("乘客已上车 {}", orderId); break; case ORDER_EN_ROUTE_TO_DESTINATION: break; case ORDER_ARRIVED_AT_DESTINATION: break; case ORDER_WAITING_FOR_DROPOFF: break; case ORDER_DROPOFF_COMPLETED: break; case ORDER_COMPLETED: break; case ORDER_CANCELLED: break; case ORDER_ARRIVED_AT_CRUISE_DESTINATION: break; case ORDER_PAUSED: break; case ORDER_EN_ROUTE_TO_CRUISE_DESTINATION: } } } } } }