中环scada增加定时迁移

bespoke-gateway-zhscada/src/main/java/com/shkpr/service/bespokegateway/zhscada/components/DataMigrator.java

@@ -21,6 +21,8 @@ import org.springframework.core.io.support.ResourcePatternResolver;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -64,8 +66,10 @@ public class DataMigrator {
 
     /**
      * 迁移scada
+     *
+     * @param previousHours 往前回溯小时数
      */
-    public void migrateScada() {
+    public void migrateScada(Integer previousHours) {
         LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
                 , "开始迁移Scada数据");
         long begin = System.currentTimeMillis();
@@ -84,18 +88,26 @@ public class DataMigrator {
                 //传感器code
                 String code = sensorMap.getOrDefault(tag.getTag(), null);
                 if (StringUtils.isBlank(code)) continue;
+
                 //数据时间
                 SensorCollectData timeRange = sensorCollectDataService.findTimeRange(migrateProperties.getSchema(), code);
                 if (Objects.isNull(timeRange) || timeRange.getMinTime() == null || timeRange.getMaxTime() == null) {
                     LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_ERROR, BIZ_TYPE, CLASS_NAME
-                            , String.format("读取Scada数据空数据, tag:%s, code: %s", tag.getTag(), code)
-                    );
+                            , String.format("读取Scada数据空数据, tag:%s, code: %s", tag.getTag(), code));
+                    continue;
+                }
+                LocalDateTime startTime = previousHours == null ? timeRange.getMinTime() : LocalDateTime.now().minusHours(previousHours).truncatedTo(ChronoUnit.HOURS);
+                LocalDateTime endTime = timeRange.getMaxTime();
+                if (startTime.isAfter(endTime)) {
+                    LogPrintMgr.getInstance().printLogMsg(LogLevelFlag.LOG_INFO, BIZ_TYPE, CLASS_NAME
+                            , String.format("开始时间大于结束时间，跳过处理, tag:%s, code:%s, startTime:%s, endTime:%s",
+                                    tag.getTag(), code, startTime, endTime));
                     continue;
                 }
 
                 //查询数据
                 List<SensorCollectData> dates = sensorCollectDataService.findAlign(migrateProperties.getSchema(),
-                        timeRange.getMinTime(), timeRange.getMaxTime(), migrateProperties.getInterval(),
+                        startTime, endTime, migrateProperties.getInterval(),
                         migrateProperties.getIntervalUnit(), migrateProperties.getAlignUnit(), code);
 
                 //构建influxdb

bespoke-gateway-zhscada/src/main/java/com/shkpr/service/bespokegateway/zhscada/manager/DataMigrateManager.java

@@ -5,6 +5,7 @@ import com.shkpr.service.bespokegateway.zhscada.components.DataMigrator;
 import com.shkpr.service.bespokegateway.zhscada.constants.ScadaPlatformMetadata;
 import com.shkpr.service.bespokegateway.zhscada.properties.MigrateProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 
@@ -39,6 +40,16 @@ public class DataMigrateManager {
     public void init() {
         //迁移scada
         if (schedulingProperties.isTaskActive(ScadaPlatformMetadata.SchedulingKeys.MIGRATE_SCADA))
-            taskScheduler.execute(dataMigrator::migrateScada);
+            taskScheduler.execute(() -> dataMigrator.migrateScada(30 * 24));
+    }
+
+    /**
+     * 小时任务
+     */
+    @Scheduled(cron = "0 * * * * ?")
+    public void hourTask() {
+        //迁移scada
+        if (schedulingProperties.isTaskActive(ScadaPlatformMetadata.SchedulingKeys.MIGRATE_SCADA))
+            taskScheduler.execute(() -> dataMigrator.migrateScada(1));
     }
 }