xp-robotaxi-automatic-pilot-boot/src/main/java/com/xiaopeng/automaticPilot/service/KafkaConsumerService.java
@@ -31,9 +31,11 @@ import com.xiaopeng.api.vo.automaticPilot.VehiclePosition; import com.xiaopeng.api.vo.automaticPilot.protobuf.*; import com.xiaopeng.api.vo.operation.RouteVo; import com.xiaopeng.api.vo.order.OrderTrajectoryVo; import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo; import com.xiaopeng.automaticPilot.util.XPUUtil; import com.xiaopeng.common.core.constant.CacheConstants; import com.xiaopeng.common.core.constant.Constants; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.enums.OrderStatusEnum; import com.xiaopeng.common.core.enums.PushMessageTypeEnum; @@ -53,6 +55,8 @@ import javax.annotation.Resource; import java.util.*; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -363,6 +367,16 @@ // dto.setDistance(rtTripInfo.getRemaining_distance().intValue()); // dto.setDirection(); xpPushMessageClient.pushOrderRoute(dto); //存储历史轨迹 OrderTrajectoryVo orderTrajectory = new OrderTrajectoryVo(); orderTrajectory.setOrderId(orderId); orderTrajectory.setLocation(longitude + "," + latitude); orderTrajectory.setDirection(dto.getDirection()); orderTrajectory.setSpeed(0D); orderTrajectory.setLocatetime(System.currentTimeMillis()); orderTrajectory.setCreateTime(new Date()); xpOrderClient.saveVehicleLocation(orderTrajectory); } } } @@ -960,8 +974,24 @@ dto.setAppUserId(appUserId + ""); dto.setMessage(message); R<String> stringR = xpPushMessageClient.sendStatus(dto); if (R.isError(stringR)) { log.error("推送消息失败：{}", stringR.getMsg()); //推送失败，重试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<String> 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()); } }); } } xp-robotaxi-operation-boot/src/main/java/com/xiaopeng/operation/utils/TaskOrderUtil.java
@@ -25,6 +25,7 @@ import com.xiaopeng.api.vo.operation.VehiclePositionVo; import com.xiaopeng.api.vo.vehicleConvoy.AddTaxiPassengerVo; 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.redis.service.RedisService; import com.xiaopeng.operation.mapper.XpElectronicFenceMapper; @@ -43,6 +44,8 @@ import javax.annotation.Resource; import java.util.*; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @@ -218,7 +221,25 @@ SendMessageDTO dto = new SendMessageDTO(); dto.setAppUserId(xpOrder.getAppUserId().toString()); dto.setMessage("{\"type\":\"ORDER_STATUS\",\"content\":\"{\\\"orderId\\\":\\\"" + orderId + "\\\",\\\"status\\\":0}\",\"timestamp\":\"" + System.currentTimeMillis() + "\"}"); xpPushMessageClient.sendStatus(dto); R<String> 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<String> 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()); } }); } //将订单从队列中移除 log.info("五轮推单结束，无车辆接单，执行队列移除订单 {}", orderNo); redisService.popFirstElement(quoteName); xp-robotaxi-order-boot/src/main/java/com/xiaopeng/order/job/jobs/EndTripJob.java
@@ -1,11 +1,13 @@ package com.xiaopeng.order.job.jobs; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONObject; import com.xiaopeng.api.dto.pushMessage.SendMessageDTO; import com.xiaopeng.api.model.order.XpOrder; import com.xiaopeng.api.model.order.XpOrderTimeoutLog; import com.xiaopeng.api.model.vehicleConvoy.XpVehicle; import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo; import com.xiaopeng.common.core.constant.Constants; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.enums.OrderStatusEnum; import com.xiaopeng.common.core.enums.PushMessageTypeEnum; @@ -18,6 +20,9 @@ import java.util.Date; import java.util.HashMap; import java.util.Map; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * 推送定时任务 @@ -51,7 +56,25 @@ SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(xpOrder.getAppUserId())); xpPushMessageClient.sendStatus(sendMessageDTO); R<String> stringR = xpPushMessageClient.sendStatus(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("定时任务结束订单后，退订订单状态消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } } log.info("结束执行自动结束行程任务"); // 添加超时记录 xp-robotaxi-order-boot/src/main/java/com/xiaopeng/order/service/impl/XpOrderServiceImpl.java
@@ -54,6 +54,7 @@ import com.xiaopeng.api.vo.vehicleConvoy.AddTaxiPassengerVo; import com.xiaopeng.api.vo.vehicleConvoy.CancelTaxiPassengerVo; import com.xiaopeng.common.core.constant.CacheConstants; import com.xiaopeng.common.core.constant.Constants; import com.xiaopeng.common.core.constant.HttpStatus; import com.xiaopeng.common.core.domain.R; import com.xiaopeng.common.core.enums.OrderStatusEnum; @@ -114,6 +115,8 @@ import java.text.SimpleDateFormat; import java.time.*; import java.util.*; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.stream.Collectors; @@ -641,7 +644,27 @@ SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(userId)); xpPushMessageClient.sendStatus(sendMessageDTO); R<String> stringR = xpPushMessageClient.sendStatus(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("运维人员取消订单推送订单状态消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } return; } // 1. 更新订单状态为已取消(0) @@ -667,7 +690,25 @@ SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(userId)); xpPushMessageClient.sendStatus(sendMessageDTO); R<String> stringR = xpPushMessageClient.sendStatus(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("运维人员取消订单推送订单状态消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } log.info("运营人员取消订单成功，推送乘客消息完成"); //查询车辆信息 @@ -730,8 +771,25 @@ SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(order.getAppUserId())); xpPushMessageClient.sendStatus(sendMessageDTO); log.info("运营人员取消订单成功，推送乘客消息完成"); R<String> stringR = xpPushMessageClient.sendStatus(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("运维人员关闭订单推送订单状态消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } log.info("订单关闭成功，订单ID: {}, 操作人: {}", order.getId(), SecurityUtils.getUsername()); } @@ -1330,7 +1388,27 @@ SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(appUserId)); xpPushMessageClient.sendStatus(sendMessageDTO); R<String> stringR = xpPushMessageClient.sendStatus(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("定时任务取消订单推送订单状态消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } log.info("定时任务取消订单推送消息完毕"); // 移除mongo数据 @@ -1425,6 +1503,39 @@ xpOrder.setTripBegin(LocalDateTime.now()); break; case 7: // 推送倒计时消息 Map<String, Object> params = new HashMap<>(); Map<String, Object> content = new HashMap<>(); content.put("orderId", xpOrder.getId().toString()); content.put("isOpen", 1); params.put("type", PushMessageTypeEnum.OPEN_TRUNK.getDesc()); params.put("content", JSONObject.toJSONString(content)); params.put("timestamp", new Date().getTime() + ""); String message = JSONObject.toJSONString(params); SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(xpOrder.getAppUserId())); R<String> stringR = xpPushMessageClient.sendStatus(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("推送倒计时消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } //创建定时结束订单和等待结束后下发巡游单的定时任务 Integer arriveTime = xpSystemSettingClient.getOrderFinishTime().getData(); // 添加自动结束行程定时任务job Map<String, ? extends Object> endTripJobMaps = @@ -1454,19 +1565,6 @@ DateUtils.localDateTimeToDate(LocalDateTime.now().plusMinutes(point.getParkingTimeLimit())), jobMaps ); // 推送倒计时消息 Map<String, Object> params = new HashMap<>(); Map<String, Object> content = new HashMap<>(); content.put("orderId", xpOrder.getId().toString()); content.put("isOpen", 1); params.put("type", PushMessageTypeEnum.OPEN_TRUNK.getDesc()); params.put("content", JSONObject.toJSONString(content)); params.put("timestamp", new Date().getTime() + ""); String message = JSONObject.toJSONString(params); SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(xpOrder.getAppUserId())); xpPushMessageClient.sendStatus(sendMessageDTO); break; } xpOrder.setStatus(status); @@ -2433,6 +2531,9 @@ public void saveVehicleLocation(OrderTrajectory orderTrajectory) { // 查询订单信息 XpOrder xpOrder = this.getById(orderTrajectory.getOrderId()); if(xpOrder.getStatus() != 6){ return; } // 计算距离 // 获取上一次定位 @@ -2715,8 +2816,26 @@ SendMessageDTO messageDTO = new SendMessageDTO(); messageDTO.setAppUserId(appUserId); messageDTO.setMessage(jsonObject.toJSONString()); System.err.println(messageDTO); xpPushMessageClient.sendRoute(messageDTO); R<String> stringR = xpPushMessageClient.sendRoute(messageDTO); XpOrder xpOrder = this.getById(orderId); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(messageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("推送乘客端导航数据消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(messageDTO), xpOrder.getOrderNumber()); } }); } } } @@ -2819,7 +2938,25 @@ SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setMessage(message); sendMessageDTO.setAppUserId(String.valueOf(appUserId)); xpPushMessageClient.sendStatus(sendMessageDTO); R<String> stringR = xpPushMessageClient.sendStatus(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("运维人员修改订单价格推送订单状态消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } log.info("订单修改价格推送消息完毕"); } } xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java
@@ -10,6 +10,7 @@ import com.xiaopeng.api.model.order.XpOrder; import com.xiaopeng.api.vo.operation.RouteVo; 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.PushMessageTypeEnum; import com.xiaopeng.common.redis.service.RedisService; @@ -25,6 +26,8 @@ import org.springframework.web.bind.annotation.RestController; import java.util.*; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @Slf4j @@ -108,7 +111,26 @@ SendMessageDTO sendMessageDTO = new SendMessageDTO(); sendMessageDTO.setAppUserId(String.valueOf(xpOrder.getAppUserId())); sendMessageDTO.setMessage(JSONObject.toJSONString(params)); xpPushMessageClient.sendRoute(sendMessageDTO); R<String> stringR = xpPushMessageClient.sendRoute(sendMessageDTO); //推送失败，重试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<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO); if(stringR1.getCode() == Constants.SUCCESS){ break; } try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } log.info("推送乘客端剩余导航数据消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); } }); } log.info("推送剩余导航数据成功 {} {}", xpOrder.getOrderNumber(), JSON.toJSONString(routeVo)); } return R.ok(); xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/utils/MapMatchingUtil.java
@@ -82,11 +82,9 @@ //判断当前位置是否处于两个点位之间 if((distance1 + distance2) == distance3 || (distance1 + distance2) <= distance3 + 5){ //取距离最近的点位数据 if(distance1 < distance2){ return polyline.subList(i, polyline.size()); } return polyline.subList(i + 1, polyline.size()); List<LonLatDto> lonLatDtos = polyline.subList(i + 1, polyline.size()); lonLatDtos.add(0, new LonLatDto(longitude, latitude)); return lonLatDtos; } } polyline.add(0, new LonLatDto(longitude, latitude));
