定时任务优化

api-admin/src/main/java/com/glxp/sale/admin/thread/HeartTask.java

@@ -22,12 +22,22 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 @Component
 @EnableScheduling
 public class HeartTask implements SchedulingConfigurer {
 
     final Logger logger = LoggerFactory.getLogger(HeartTask.class);
+    
+    // 使用与HeartTaskService一致的常量名
+    private static final String REDIS_DOWNLOAD_STATUS_KEY = "SC_UDIINFO_DOWNLOAD_STATUS";
+    
+    // 用于确保单个JVM实例内的并发安全
+    private final Lock processLock = new ReentrantLock();
+    
     @Resource
     protected ScheduledDao scheduledDao;
     @Resource
@@ -41,42 +51,91 @@ public class HeartTask implements SchedulingConfigurer {
 
     @Override
     public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
-
         scheduledTaskRegistrar.addTriggerTask(() -> process(), triggerContext -> {
+            try {
                 ScheduledRequest scheduledRequest = new ScheduledRequest();
                 scheduledRequest.setCronName("heartTask");
                 ScheduledEntity scheduledEntity = scheduledDao.findScheduled(scheduledRequest);
+                
+                // 健壮性检查cron表达式
+                if (scheduledEntity == null) {
+                    logger.error("未找到heartTask的计划任务配置");
+                    return new CronTrigger("0 0/10 * * * ?").nextExecutionTime(triggerContext);
+                }
+                
                 String cron = scheduledEntity.getCron();
-            if (cron.isEmpty()) {
+                if (cron == null || cron.isEmpty()) {
-                logger.error("cron is null");
+                    logger.error("cron表达式为空，使用默认配置");
+                    return new CronTrigger("0 0/10 * * * ?").nextExecutionTime(triggerContext);
                 }
+                
                 return new CronTrigger(cron).nextExecutionTime(triggerContext);
+            } catch (Exception e) {
+                logger.error("获取cron表达式异常，使用默认配置", e);
+                return new CronTrigger("0 0/10 * * * ?").nextExecutionTime(triggerContext);
+            }
         });
     }
 
-
     private void process() {
+        boolean lockAcquired = false;
+        try {
+            // 尝试获取锁，避免同一JVM内多线程同时执行
+            lockAcquired = processLock.tryLock(5, TimeUnit.SECONDS);
+            if (!lockAcquired) {
+                logger.warn("心跳任务锁获取失败，可能有其他线程正在执行");
+                return;
+            }
+
+            logger.info("------------心跳任务开始-----------------");
             
-        logger.info("------------心跳任务-----------------");
             SystemParamConfigEntity systemParamConfigEntity = systemParamConfigService.selectByParamKey("sc_udiinfo_status");
-        long timeInterval = Long.parseLong(systemParamConfigEntity.getParamValue()) * 6 * 1000;
+            if (systemParamConfigEntity == null || systemParamConfigEntity.getParamValue() == null) {
+                logger.error("系统参数sc_udiinfo_status未配置");
+                return;
+            }
+            
+            try {
+                // 确保使用正确的乘数 10
+                long timeInterval = Long.parseLong(systemParamConfigEntity.getParamValue()) * 10 * 1000;
                 long curTime = System.currentTimeMillis();
-        Long lastTime = (Long) redisUtil.get("SC_UDIINFO_DOWNLOAD_STATUS");
+                Long lastTime = (Long) redisUtil.get(REDIS_DOWNLOAD_STATUS_KEY);
+                
                 if (lastTime == null) {
                     try {
                         lastTime = DateUtil.timeToStamp("1949-01-01 00:00:00");
                     } catch (Exception e) {
-                e.printStackTrace();
+                        logger.error("时间戳转换失败", e);
+                        lastTime = 0L; // 使用安全默认值
                     }
-            redisUtil.set("SC_UDIINFO_DOWNLOAD_STATUS", lastTime);
+                    // 首次初始化Redis时间戳
+                    redisUtil.set(REDIS_DOWNLOAD_STATUS_KEY, lastTime);
                 }
+                
                 if (curTime - lastTime > timeInterval) {
+                    logger.info("时间间隔已满足，开始执行数据下载");
+                    
+                    // 直接调用heartTaskService.dlAllData()
+                    // 注意：不在此处更新REDIS_DOWNLOAD_STATUS_KEY，避免与HeartTaskService的冲突
+                    // 由HeartTaskService.dlAllData()负责更新REDIS_DOWNLOAD_STATUS_KEY
                     heartTaskService.dlAllData();
                 } else {
-            logger.info("时间未到");
+                    logger.info("时间未到，距离下次执行还有{}毫秒", timeInterval - (curTime - lastTime));
                 }
-
+            } catch (NumberFormatException e) {
+                logger.error("解析时间间隔参数失败: {}", systemParamConfigEntity.getParamValue(), e);
+            } catch (Exception e) {
+                logger.error("心跳任务执行异常", e);
             }
             
-
+            logger.info("------------心跳任务结束-----------------");
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            logger.error("获取心跳任务锁时被中断", e);
+        } finally {
+            if (lockAcquired) {
+                processLock.unlock();
+            }
+        }
+    }
 }

api-admin/src/main/java/com/glxp/sale/admin/thread/HeartTaskService.java

@@ -14,11 +14,20 @@ import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
 import java.util.Arrays;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 
 @Service
 public class HeartTaskService {
 
-    final Logger logger = LoggerFactory.getLogger(HeartTask.class);
+    private static final Logger logger = LoggerFactory.getLogger(HeartTaskService.class);
+    // 确保常量命名与HeartTask保持一致
+    private static final String REDIS_DOWNLOAD_STATUS_KEY = "SC_UDIINFO_DOWNLOAD_STATUS";
+    private static final String REDIS_UPLOAD_STATUS_KEY = "UPLOAD_UDIINFO_STATUS";
+    private static final String DOWNLOAD_LOCK_KEY = "is_doing_download";
+    // Redis锁过期时间(秒)：设置合理的超时时间，避免任务卡死导致锁无法释放
+    private static final int LOCK_EXPIRY_SECONDS = 30 * 60; // 30分钟
+
     @Resource
     SystemParamConfigService systemParamConfigService;
     @Resource
@@ -34,91 +43,85 @@ public class HeartTaskService {
     @Value("${SPMS_WEBSOCKET_TOKEN}")
     private String socketToken;
 
-    //定时从上游下载数据
-    public void dlData() {
-        SystemParamConfigEntity upConnect = systemParamConfigService.selectByParamKey("sync_upstream_enable");
-        if (upConnect != null && upConnect.getParamValue().equals("1")) {
-            dlAllData();
-            scanUpload();
-        }
-        SystemParamConfigEntity donwConnect = systemParamConfigService.selectByParamKey("sync_downstream_enable");
-        if (donwConnect != null && donwConnect.getParamValue().equals("1")) {
-            scanDonwload();
-        }
-    }
 
-    private void scanUpload() {
+    /**
-        SystemParamConfigEntity systemParamConfigEntity = systemParamConfigService.selectByParamKey("sc_udiinfo_upload");
+     * 从上游下载全部数据
-        long timeInterval = Long.parseLong(systemParamConfigEntity.getParamValue()) * 60 * 1000;
+     * 该方法支持被HeartTask或其他服务直接调用
-        long curTime = System.currentTimeMillis();
+     * 内部实现了基于Redis的分布式锁，确保在分布式环境下也只有一个实例执行
-        //定时扫描
+     */
-        Long lastTime = (Long) redisUtil.get("UPLOAD_UDIINFO_STATUS");
+    public void dlAllData() {
-        if (lastTime == null) {
+        logger.info("定时从上游下载全部据-----开始");
-            lastTime = System.currentTimeMillis();
+        // 先更新时间戳，防止其他实例重复执行
-            redisUtil.set("UPLOAD_UDIINFO_STATUS", lastTime, 30 * 60);
+        // 此处的时间戳更新会被HeartTask及其他服务读取用于判断是否执行
-        }
+        redisUtil.set(REDIS_DOWNLOAD_STATUS_KEY, System.currentTimeMillis());
-        if (curTime - lastTime > timeInterval) {
-            redisUtil.set("UPLOAD_UDIINFO_STATUS", curTime);
-            scanUploadService.scanAllDatas();
-            scanUploadService.scanAllBus();
-            scanUploadService.scanAllOrders();
-            scanUploadService.scanAllSchedule();
-        }
-    }
 
-    private void scanDonwload() {
+        // 生成唯一的锁标识，用于安全释放锁（避免释放其他实例的锁）
-        SystemParamConfigEntity upConnect = systemParamConfigService.selectByParamKey("sync_downstream_enable");
+        String lockValue = UUID.randomUUID().toString();
-        if (upConnect.getParamValue().equals("1")) {
+        boolean lockAcquired = false;
-            SystemParamConfigEntity systemParamConfigEntity = systemParamConfigService.selectByParamKey("sc_udiinfo_status");
-            long timeInterval = Long.parseLong(systemParamConfigEntity.getParamValue()) * 60 * 1000;
-            long curTime = System.currentTimeMillis();
-            Long lastTime = (Long) redisUtil.get("SC_UDIINFO_DOWNLOAD_STATUS");
-            if (lastTime == null) {
-                lastTime = System.currentTimeMillis();
-                redisUtil.set("SC_UDIINFO_DOWNLOAD_STATUS", lastTime);
-            }
-            if (curTime - lastTime > timeInterval) {
-                redisUtil.set("SC_UDIINFO_DOWNLOAD_STATUS", curTime);
-                scanDownloadService.scanAllData();
-                scanDownloadService.scanAllBus();
-                scanDownloadService.scanAllOrder();
-                scanDownloadService.scanScheduleList();
-                scanDownloadService.scanUdis();
-            }
-        }
 
+        try {
+            // 检查是否已有下载任务在进行
+            String existingLock = (String) redisUtil.get(DOWNLOAD_LOCK_KEY);
+            if (existingLock != null && !existingLock.equals("false")) {
+                logger.info("已有下载任务正在执行中，跳过本次执行");
+                return;
             }
 
-    public void dlUpAllData() {
+            // 尝试获取锁，使用唯一值标识锁的拥有者
-        SystemParamConfigEntity upConnect = systemParamConfigService.selectByParamKey("sync_upstream_enable");
+            lockAcquired = redisUtil.set(DOWNLOAD_LOCK_KEY, lockValue, LOCK_EXPIRY_SECONDS);
-        if (upConnect != null && upConnect.getParamValue().equals("1")) {
+            if (!lockAcquired) {
-            dlAllData();
+                logger.warn("无法获取下载任务锁，可能有其他进程正在执行");
-        }
+                return;
             }
-
+            logger.info("成功获取下载锁，开始执行数据下载任务，锁ID: {}", lockValue);
-
+            try {
-    public void dlAllData() {
+                // 执行订单数据下载
-        logger.info("定时从上游下载全部据-----");
-        redisUtil.set("SC_UDIINFO_DOWNLOAD_STATUS", System.currentTimeMillis());
-        String doing = (String) redisUtil.get("is_doing_download");
-        if (doing == null || doing.equals("false")) {
-            redisUtil.set("is_doing_download", "true", 60);
                 dlAllDataService.dllNewAllOrder();
+                // 执行其他类型数据下载
                 Arrays.stream(BasicExportTypeEnum.values()).forEach(i -> {
+                    try {
                         dlAllDataService.pullData(i);
+                    } catch (Exception e) {
+                        logger.error("下载数据类型[{}]时发生异常", i, e);
+                    }
                 });
-            dlAllDataService.dlAllDi();
+                // 注释掉的DI下载，保留注释
-            redisUtil.set("is_doing_download", "false");
+                // dlAllDataService.dlAllDi();
-            spsSyncWebSocket.sendMessage(SocketMsgEntity.builder().type(SocketMsgType.DL_ALL_DATA).content("").remark("下载基础信息").build(), "2:" + socketToken);
+                // 发送WebSocket消息通知
-
+                try {
+                    spsSyncWebSocket.sendMessage(
+                            SocketMsgEntity.builder()
+                                    .type(SocketMsgType.DL_ALL_DATA)
+                                    .content("")
+                                    .remark("下载基础信息")
+                                    .build(),
+                            "2:" + socketToken
+                    );
+                } catch (Exception e) {
+                    logger.error("发送WebSocket消息失败", e);
+                }
+                logger.info("数据下载任务执行完成");
+            } catch (Exception e) {
+                logger.error("执行数据下载任务过程中发生异常", e);
+            }
+        } finally {
+            // 确保无论如何都释放锁，但只释放自己的锁
+            if (lockAcquired) {
+                // 检查锁是否仍然是我们设置的值，避免释放他人的锁
+                String currentLock = (String) redisUtil.get(DOWNLOAD_LOCK_KEY);
+                if (lockValue.equals(currentLock)) {
+                    boolean released = redisUtil.set(DOWNLOAD_LOCK_KEY, "false");
+                    if (!released) {
+                        logger.error("释放下载任务锁失败");
+                    } else {
+                        logger.info("成功释放下载锁，锁ID: {}", lockValue);
+                    }
+                } else {
+                    logger.warn("锁已被其他进程修改，不进行释放，当前锁值: {}, 期望锁值: {}", currentLock, lockValue);
+                }
             }
             logger.info("定时从上游下载全部据-----结束");
-
         }
-
-
-    public void getData(String message){
-        logger.error("jiehsou 要发送的数据");
-        spsSyncWebSocket.sendMessage(SocketMsgEntity.builder().content(message).type(SocketMsgType.STAT_DATA).remark("我瞎说的数据").build(), "2:" + socketToken);
     }
+
 }