From 0bcfca6b2561d096a2ce26e2e7200cd0768b2b6d Mon Sep 17 00:00:00 2001
From: liujie <1793218484@qq.com>
Date: 星期二, 12 五月 2026 09:27:03 +0800
Subject: [PATCH] Merge branch '1.0' of http://120.76.84.145:10101/gitblit/r/java/Robotaxi into 1.0

---
 xp-robotaxi-push-message-boot/src/main/java/com/xiaopeng/pushMessage/controller/PushOrderRouteController.java |   38 +++++++++++++++++++++++++++++++++++---
 1 files changed, 35 insertions(+), 3 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 3edc4fe..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
@@ -26,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;
@@ -89,6 +119,7 @@
 
                 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));
@@ -113,6 +144,7 @@
 
                 // 封装路由详情
                 Map<String, Object> route = new HashMap<>();
+                route.put("speed",speed);
                 route.put("distance",distance);
                 route.put("duration",duration);
                 route.put("polyline",JSONObject.toJSONString(lonLatDtos));
@@ -139,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){
@@ -149,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