From 08cc6d5cf7e93b75cfabe7dcc2aeaa9afa1117eb Mon Sep 17 00:00:00 2001
From: puzhibing <393733352@qq.com>
Date: 星期二, 12 五月 2026 15:01:47 +0800
Subject: [PATCH] Merge branch '1.0' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi into 1.1
---
xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java | 102 +++++++++++++++++++++++++++++++++++----------------
1 files changed, 70 insertions(+), 32 deletions(-)
diff --git a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java b/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java
index d77157d..3cb5a77 100644
--- a/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java
+++ b/xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java
@@ -6,20 +6,17 @@
import com.xiaopeng.api.dto.pushMessage.PushOrderRouteDTO;
import com.xiaopeng.api.dto.pushMessage.SendMessageDTO;
import com.xiaopeng.api.feignClient.order.XpOrderClient;
-import com.xiaopeng.api.feignClient.order.XpTestOrderClient;
import com.xiaopeng.api.feignClient.pushMessage.XpPushMessageClient;
import com.xiaopeng.api.model.order.XpOrder;
-import com.xiaopeng.api.model.order.XpTestOrder;
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;
import com.xiaopeng.pushMessage.utils.MapMatchingUtil;
+import com.xiaopeng.pushMessage.utils.TrajectoryDensifier;
import io.swagger.annotations.Api;
import lombok.extern.slf4j.Slf4j;
-import org.powermock.core.ListMap;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
@@ -29,14 +26,44 @@
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
@Slf4j
@Api(tags = "订单推送轨迹接口")
@RestController
@RequestMapping("")
public class PushOrderRouteController {
+
+ /**
+ * 推送轨迹重试线程池（类级别单例复用）
+ * 修复：原代码在每次 sendRoute 失败时都 new ThreadPoolExecutor 且不 shutdown，
+ * 导致核心线程在 LinkedBlockingQueue.take() 上永久 WAITING，造成线程泄漏。
+ *
+ * 参数推导（基于业务并发量估算）：
+ * - 车端上报频率 ~1次/秒/车，接口限频1.5秒，实际处理 ~0.67次/秒/车
+ * - 常态（50车 + 5%失败率）：~1.65 任务/秒，任务耗时~5秒，同时~8个任务
+ * - 突发（100车 + 20%失败率）：~13 任务/秒，同时~66个任务
+ * - corePoolSize=4 覆盖常态，maxPoolSize=16 + 队列512 覆盖突发
+ */
+ private static final ThreadPoolExecutor RETRY_EXECUTOR = new ThreadPoolExecutor(
+ 4,
+ 16,
+ 60L, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(512),
+ new ThreadFactory() {
+ private final AtomicInteger counter = new AtomicInteger(0);
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(r, "route-push-retry-" + counter.incrementAndGet());
+ t.setDaemon(true);
+ return t;
+ }
+ },
+ new ThreadPoolExecutor.CallerRunsPolicy()
+ );
@Autowired
private RedisService redisService;
@@ -55,23 +82,47 @@
Integer routeType = dto.getRouteType();
//正常订单
if(routeType == 1){
+ //非导航偏离才需要限频
+ if(PushMessageTypeEnum.ORDER_ROUTE.getDesc().equals(dto.getMessageType())){
+ Long pushTime = redisService.getCacheObject("PUSH_TIME:" + dto.getOrderId());
+ if(null != pushTime && System.currentTimeMillis() - pushTime < 1500){
+ return R.ok();
+ }
+ redisService.setCacheObject("PUSH_TIME:" + dto.getOrderId(), System.currentTimeMillis());
+ }
XpOrder xpOrder = xpOrderClient.queryOrderById(dto.getOrderId()).getData();
+ if(xpOrder.getStatus() <= 1 || xpOrder.getStatus() >= 8){
+ return R.ok();
+ }
// 获取路线规划
String key = "";
//接驾情况，获取车辆位置到起点的导航
- if(xpOrder.getStatus() == 2 || xpOrder.getStatus() == 3){
+ if(xpOrder.getStatus() >= 2 && xpOrder.getStatus() <= 4){
key = "ORDER_NAVIGATION_START_DATA:" + xpOrder.getAppUserId();
}else{
key = "ORDER_NAVIGATION_END_DATA:" + xpOrder.getAppUserId();
}
Object cacheObject = redisService.getCacheObject(key);
RouteVo routeVo = JSON.parseObject(cacheObject.toString(), RouteVo.class);
-
+ log.info("缓存获取的轨迹 {} {}", key, JSON.toJSONString(routeVo));
if(Objects.nonNull(routeVo)){
- Integer distance = dto.getDistance();
- Integer duration = dto.getDuration();
+ //开始对轨迹进行点位补全，然后更新缓存中的轨迹数据
+ if(null == routeVo.getDensify() || routeVo.getDensify() == 0){
+ log.info("开始执行点位补全 {}", key);
+ List<LonLatDto> polyline = routeVo.getPolyline();
+ List<LonLatDto> densify = TrajectoryDensifier.densify(polyline, 15);
+ routeVo.setPolyline(densify);
+ routeVo.setDensify(1);
+ log.info("点位补全结束 {} {}", key, JSON.toJSONString(routeVo));
+ redisService.setCacheObject(key, JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
+ }
+
+ Integer distance = dto.getDistance().intValue();
+ Integer duration = dto.getDuration().intValue();
+ Integer speed = dto.getDuration().intValue();
// 计算实时点位到线段的最优截取
Map<String, List<LonLatDto>> remainingNavigation = MapMatchingUtil.getRemainingNavigation(routeVo.getPolyline(), dto.getLongitude(), dto.getLatitude());
+ log.info("remainingNavigation {}", JSON.toJSONString(remainingNavigation));
//当订单状态为5的时候（乘客已上车）需要推送导航的距离和剩余时间
//因为这个状态，车端推的依然是接驾的剩余里程和时间，就会导致数据和小程序的页面跳转匹配不上
//小程序是只要乘客上车后就跳转到行程开始页面
@@ -84,45 +135,32 @@
}
List<LonLatDto> lonLatDtos = remainingNavigation.get("unserved");
List<LonLatDto> lonLatDtos1 = remainingNavigation.get("served");
+
routeVo.setPolyline(lonLatDtos);
routeVo.setServed(lonLatDtos1);
-
+ log.info("线路截取后存储的轨迹 {}", JSON.toJSONString(routeVo));
// 存储最新路线偏离缓存，后续进行轨迹截取
redisService.setCacheObject(key, JSON.toJSONString(routeVo), 24L, TimeUnit.HOURS);
- //将车辆的当前位置加入到第一个轨迹点位集合中
- lonLatDtos.add(0, new LonLatDto(dto.getLongitude(), dto.getLatitude()));
- // 封装轨迹点位集合
- List<Map<String, Object>> polyline = new ArrayList<>();
- for (LonLatDto lonLatDto : lonLatDtos) {
- Map<String, Object> polylineMap = new HashMap<>();
- polylineMap.put("longitude",lonLatDto.getLongitude());
- polylineMap.put("latitude",lonLatDto.getLatitude());
- polyline.add(polylineMap);
- }
- List<Map<String, Object>> served = new ArrayList<>();
- for (LonLatDto lonLatDto : lonLatDtos1) {
- Map<String, Object> polylineMap = new HashMap<>();
- polylineMap.put("longitude",lonLatDto.getLongitude());
- polylineMap.put("latitude",lonLatDto.getLatitude());
- served.add(polylineMap);
- }
// 封装路由详情
Map<String, Object> route = new HashMap<>();
+ route.put("speed",speed);
route.put("distance",distance);
route.put("duration",duration);
- route.put("polyline",JSONObject.toJSONString(polyline));
- route.put("served",JSONObject.toJSONString(served));
+ route.put("polyline",JSONObject.toJSONString(lonLatDtos));
+ route.put("served",JSONObject.toJSONString(lonLatDtos1));
Map<String, Object> content = new HashMap<>();
content.put("orderId",dto.getOrderId().toString());
content.put("routeType",routeType);
content.put("direction",dto.getDirection());
content.put("route",JSONObject.toJSONString(route));
+ content.put("endLongitude", dto.getEndLongitude());
+ content.put("endLatitude", dto.getEndLatitude());
// 封装轨迹信息
Map<String, Object> params = new HashMap<>();
- params.put("type", PushMessageTypeEnum.ORDER_ROUTE.getDesc());
+ params.put("type", dto.getMessageType());
params.put("content", JSONObject.toJSONString(content));
params.put("timestamp", new Date().getTime()+"");
@@ -133,8 +171,7 @@
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(() -> {
+ RETRY_EXECUTOR.execute(() -> {
for (int i = 0; i < 3; i++) {
R<String> stringR1 = xpPushMessageClient.sendStatus(sendMessageDTO);
if(stringR1.getCode() == Constants.SUCCESS){
@@ -143,7 +180,8 @@
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
- throw new RuntimeException(e);
+ Thread.currentThread().interrupt();
+ return;
}
log.info("推送乘客端剩余导航数据消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber());
}
--
Gitblit v1.7.1