java/Robotaxi.git

			@@ -14,7 +14,6 @@
			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.TrackPoint;
			import com.xiaopeng.pushMessage.utils.TrajectoryDensifier;
			import io.swagger.annotations.Api;
			import lombok.extern.slf4j.Slf4j;
			@@ -25,17 +24,46 @@
			import org.springframework.web.bind.annotation.RequestMapping;
			import org.springframework.web.bind.annotation.RestController;

			import java.time.Instant;
			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;
			@@ -54,11 +82,14 @@
			Integer routeType = dto.getRouteType();
			//正常订单
			if(routeType == 1){
			Long pushTime = redisService.getCacheObject("PUSH_TIME:" + dto.getOrderId());
			if(null != pushTime && System.currentTimeMillis() - pushTime < 1500){
			return R.ok();
			//非导航偏离才需要限频
			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());
			}
			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();
			@@ -66,7 +97,7 @@
			// 获取路线规划
			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();
			@@ -88,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));
			@@ -112,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));
			@@ -122,10 +155,12 @@
			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()+"");

			@@ -136,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){
			@@ -146,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());
			}