| | |
|---|
| | | |
|---|
| | | 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; |
|---|
| | |
|---|
| | | 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){ |
|---|
| | |
|---|
| | | try { |
|---|
| | | Thread.sleep(1000); |
|---|
| | | } catch (InterruptedException e) { |
|---|
| | | throw new RuntimeException(e); |
|---|
| | | Thread.currentThread().interrupt(); |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | log.info("推送乘客端剩余导航数据消息失败：{} {} {}", stringR1.getMsg(), JSON.toJSONString(sendMessageDTO), xpOrder.getOrderNumber()); |
|---|
| | | } |
|---|
