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 | 36 +++++++++++++++++++++++++++++++++---
1 files changed, 33 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 18f5b7f..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;
@@ -141,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){
@@ -151,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