fix1209

cloud-consume/cloud-consume-biz/src/main/java/com/lutao/cloud/consume/factory/FireParametersTagStrategy.java

@@ -7,23 +7,20 @@ import com.lutao.cloud.collect.vo.DataAddVO;
 import com.lutao.cloud.common.core.kafka.MessageCommandEnum;
 import com.lutao.cloud.common.core.util.CollectionUtil;
 import com.lutao.cloud.common.core.util.MessageInfo;
-import com.lutao.cloud.common.core.util.ObjectUtil;
-import com.lutao.cloud.common.core.util.StringUtil;
 import com.lutao.cloud.consume.service.FireParametersTagService;
 import com.lutao.cloud.consume.util.TimeUtil;
 import com.lutao.entity.FireParametersTag;
 import com.lutao.entity.PerceptionMonitorFireInfo;
+import com.lutao.entity.PerceptionMonitorGasInfo;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.BoundHashOperations;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
-import org.springframework.transaction.annotation.Transactional;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
-import java.time.LocalTime;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.List;
@@ -152,41 +149,55 @@ public class FireParametersTagStrategy implements CommandStrategy<FireParameters
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void stopTag(String appId, String platformCode, String tagCode) {
+    private void stopTagCache(String appId, String platformCode, String tagCode) {
         log.info("fireParametersTagStrategy 进行stopTag操作 ......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
         BoundHashOperations parametersTagCache = fireParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorFireInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count)) {
-            parametersTagCache.put("end_time", LocalDateTime.now().toString());
-            // 计算每分钟应该接受的数据量
-            LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-            Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
-            parametersTagCache.put("should_accept_count", shouldAcceptCount);
-            parametersTagCache.put("enable", 1);
+        // 2024/11/14 删除时候-判断缓存是否存在
+        Boolean hasKey = fireParametersTagStrategy.redisTemplate.hasKey(LocalDate.now() + ":" + PerceptionMonitorFireInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        if (Boolean.FALSE.equals(hasKey)){
+            return;
         }
+
+        //这个指标信息已经存在 且收到信息的情况
+        parametersTagCache.put("end_time", LocalDateTime.now().toString());
+        // 计算每分钟应该接受的数据量
+        LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
+        Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
+        parametersTagCache.put("should_accept_count", shouldAcceptCount);
+        parametersTagCache.put("enable", 1);
     }
 
-    private void startTag(String appId, String platformCode, String tagCode) {
+    private void startTagCache(FireParametersTag fireParametersTag) {
         log.info("fireParametersTagStrategy 进行startTag操作 ......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
-        BoundHashOperations parametersTagCache = fireParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorFireInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        BoundHashOperations parametersTagCache = fireParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" +
+                PerceptionMonitorFireInfo.TABLE_NAME + ":" +
+                fireParametersTag.getAppId() + ":" +
+                fireParametersTag.getPlatformCode() + ":" +
+                fireParametersTag.getTagCode());
         Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count) && count < 1) {
+        if (!Objects.nonNull(count)) {
+            // 初始化
             parametersTagCache.put("should_accept_count", 0);
             parametersTagCache.put("received_count", 0);
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.put("enable", 0);
-
+            parametersTagCache.put("resource", PerceptionMonitorFireInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(fireParametersTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
         } else {
-            //有的情况下
+            // 恢复缓存
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.delete("end_time");
             parametersTagCache.put("enable", 0);
+            parametersTagCache.put("resource", PerceptionMonitorFireInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(fireParametersTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
         }
     }
 
@@ -197,19 +208,17 @@ public class FireParametersTagStrategy implements CommandStrategy<FireParameters
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void addTag(String appId, String platformCode, String tagCode) {
+    private void addTagCache(String appId, String platformCode, String tagCode) {
         log.info("fireParametersTagStrategy 进行addTag操作 ......");
         BoundHashOperations parametersTagCache = fireParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorFireInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count)) {
-            parametersTagCache.put("begin_time", LocalDateTime.now().toString());
-            parametersTagCache.put("resource", PerceptionMonitorFireInfo.TABLE_NAME);
-            parametersTagCache.put("received_count", 0);
-            parametersTagCache.put("should_accept_count", 0);
-            parametersTagCache.put("enable", 0);
-            parametersTagCache.expire(48, TimeUnit.HOURS);
-        }
+        parametersTagCache.put("begin_time", LocalDateTime.now().toString());
+        parametersTagCache.put("resource", PerceptionMonitorFireInfo.TABLE_NAME);
+        parametersTagCache.put("received_count", 0);
+        parametersTagCache.put("should_accept_count", 0);
+        parametersTagCache.put("enable", 0);
+        parametersTagCache.expire(48, TimeUnit.HOURS);
+
     }
 
     /**
@@ -222,10 +231,9 @@ public class FireParametersTagStrategy implements CommandStrategy<FireParameters
     public void add(FireParametersTag fireParametersTag) {
         log.info("fireParametersTagStrategy 进行add操作 ......");
         fireParametersTagStrategy.fireParametersTagService.save(fireParametersTag);
-        addTag(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
         //指标停用
-        if ("1".equals(fireParametersTag.getStatus())) {
-            stopTag(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
+        if (!"1".equals(fireParametersTag.getStatus())) {
+            addTagCache(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
         }
     }
 
@@ -245,13 +253,19 @@ public class FireParametersTagStrategy implements CommandStrategy<FireParameters
         fireParametersTagStrategy.fireParametersTagService.update(fireParametersTag, Wrappers.lambdaQuery(FireParametersTag.class)
                 .eq(FireParametersTag::getAppId, fireParametersTag.getAppId())
                 .eq(FireParametersTag::getOriginId, fireParametersTag.getOriginId()));
-        // 指标停用
+        /**
+         * 1.当前状态有效0，新指标状态有效0，不操作缓存；
+         * 2.当前状态有效0，新指标状态无效1，停止缓存；->stopTagCache()
+         * 3.当前状态无效1，新指标状态有效0，恢复缓存；->startTagCache()
+         * 4.当前状态无效1，新指标状态无效1，不操作缓存；
+         */
+        // 指标缓存停用
         if ("1".equals(fireParametersTag.getStatus()) && !"1".equals(fireParametersTagTemp.getStatus())) {
-            stopTag(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
+            stopTagCache(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
         }
-        // 启用
+        // 指标缓存启用
         if ("0".equals(fireParametersTag.getStatus()) && ("1".equals(fireParametersTagTemp.getStatus()))) {
-            startTag(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
+            startTagCache(fireParametersTag);
         }
     }
 
@@ -269,7 +283,7 @@ public class FireParametersTagStrategy implements CommandStrategy<FireParameters
                 .eq(FireParametersTag::getOriginId, fireParametersTag.getOriginId()));
 
         //指标停用
-        stopTag(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
+        stopTagCache(fireParametersTag.getAppId(), fireParametersTag.getPlatformCode(), fireParametersTag.getTagCode());
     }
 
     /**

cloud-consume/cloud-consume-biz/src/main/java/com/lutao/cloud/consume/factory/GasParametersTagStrategy.java

@@ -9,10 +9,7 @@ import com.lutao.cloud.common.core.util.CollectionUtil;
 import com.lutao.cloud.common.core.util.MessageInfo;
 import com.lutao.cloud.consume.service.GasParametersTagService;
 import com.lutao.cloud.consume.util.TimeUtil;
-import com.lutao.entity.FireParametersTag;
-import com.lutao.entity.GasParametersTag;
-import com.lutao.entity.PerceptionMonitorFireInfo;
-import com.lutao.entity.PerceptionMonitorGasInfo;
+import com.lutao.entity.*;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.BoundHashOperations;
 import org.springframework.data.redis.core.RedisTemplate;
@@ -153,40 +150,53 @@ public class GasParametersTagStrategy implements CommandStrategy<GasParametersTa
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void stopTag(String appId, String platformCode, String tagCode) {
+    private void stopTagCache(String appId, String platformCode, String tagCode) {
         log.info("gasParametersTagStrategy 进行stopTag操作 .......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
         BoundHashOperations parametersTagCache = gasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count)) {
-            parametersTagCache.put("end_time", LocalDateTime.now().toString());
-            // 计算每分钟应该接受的数据量
-            LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-            Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
-            parametersTagCache.put("should_accept_count", shouldAcceptCount);
-            parametersTagCache.put("enable", 1);
+        // 2024/11/14 删除时候-判断缓存是否存在
+        Boolean hasKey = gasParametersTagStrategy.redisTemplate.hasKey(LocalDate.now() + ":" + PerceptionMonitorGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        if (Boolean.FALSE.equals(hasKey)){
+            return;
         }
+        parametersTagCache.put("end_time", LocalDateTime.now().toString());
+        // 计算每分钟应该接受的数据量
+        LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
+        Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
+        parametersTagCache.put("should_accept_count", shouldAcceptCount);
+        parametersTagCache.put("enable", 1);
     }
 
-    private void startTag(String appId, String platformCode, String tagCode) {
+    private void startTagCache(GasParametersTag gasParametersTag) {
         log.info("gasParametersTagStrategy 进行startTag操作 ......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
-        BoundHashOperations parametersTagCache = gasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        BoundHashOperations parametersTagCache = gasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":"
+                + PerceptionMonitorGasInfo.TABLE_NAME + ":"
+                + gasParametersTag.getAppId() + ":"
+                + gasParametersTag.getPlatformCode() + ":"
+                + gasParametersTag.getTagCode());
         Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count) && count < 1) {
+        if (!Objects.nonNull(count)) {
+            // 初始化
             parametersTagCache.put("should_accept_count", 0);
             parametersTagCache.put("received_count", 0);
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.put("enable", 0);
-        }else {
-            //有的情况下
+            parametersTagCache.put("resource", PerceptionMonitorGasInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(gasParametersTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
+        } else {
+            // 恢复缓存
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.delete("end_time");
             parametersTagCache.put("enable", 0);
+            parametersTagCache.put("resource", PerceptionMonitorGasInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(gasParametersTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
         }
     }
 
@@ -197,19 +207,16 @@ public class GasParametersTagStrategy implements CommandStrategy<GasParametersTa
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void addTag(String appId, String platformCode, String tagCode) {
+    private void addTagCache(String appId, String platformCode, String tagCode) {
         log.info("gasParametersTagStrategy 进行addTag操作 .......");
         BoundHashOperations parametersTagCache = gasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count)) {
-            parametersTagCache.put("begin_time", LocalDateTime.now().toString());
-            parametersTagCache.put("resource", PerceptionMonitorGasInfo.TABLE_NAME);
-            parametersTagCache.put("received_count", 0);
-            parametersTagCache.put("should_accept_count", 0);
-            parametersTagCache.put("enable", 0);
-            parametersTagCache.expire(48, TimeUnit.HOURS);
-        }
+        parametersTagCache.put("begin_time", LocalDateTime.now().toString());
+        parametersTagCache.put("resource", PerceptionMonitorGasInfo.TABLE_NAME);
+        parametersTagCache.put("received_count", 0);
+        parametersTagCache.put("should_accept_count", 0);
+        parametersTagCache.put("enable", 0);
+        parametersTagCache.expire(48, TimeUnit.HOURS);
     }
 
     /**
@@ -222,9 +229,8 @@ public class GasParametersTagStrategy implements CommandStrategy<GasParametersTa
     private void add(GasParametersTag gasParametersTag) {
         log.info("gasParametersTagStrategy 进行add操作 ......");
         gasParametersTagStrategy.gasParametersTagService.save(gasParametersTag);
-        addTag(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
-        if ("1".equals(gasParametersTag.getStatus())) {
-            stopTag(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
+        if (!"1".equals(gasParametersTag.getStatus())) {
+            addTagCache(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
         }
 
     }
@@ -245,13 +251,19 @@ public class GasParametersTagStrategy implements CommandStrategy<GasParametersTa
         gasParametersTagStrategy.gasParametersTagService.update(gasParametersTag, Wrappers.lambdaQuery(GasParametersTag.class)
                 .eq(GasParametersTag::getAppId, gasParametersTag.getAppId())
                 .eq(GasParametersTag::getOriginId, gasParametersTag.getOriginId()));
-        // 指标停用
+        /**
+         * 1.当前状态有效0，新指标状态有效0，不操作缓存；
+         * 2.当前状态有效0，新指标状态无效1，停止缓存；->stopTagCache()
+         * 3.当前状态无效1，新指标状态有效0，恢复缓存；->startTagCache()
+         * 4.当前状态无效1，新指标状态无效1，不操作缓存；
+         */
+        // 指标缓存停用
         if ("1".equals(gasParametersTag.getStatus()) && !"1".equals(gasParametersTagTemp.getStatus())) {
-            stopTag(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
+            stopTagCache(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
         }
-        // 启用
+        // 指标缓存启用
         if ("0".equals(gasParametersTag.getStatus()) && ("1".equals(gasParametersTagTemp.getStatus()))) {
-            startTag(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
+            startTagCache(gasParametersTag);
         }
     }
 
@@ -269,7 +281,7 @@ public class GasParametersTagStrategy implements CommandStrategy<GasParametersTa
                 .eq(GasParametersTag::getOriginId, gasParametersTag.getOriginId()));
 
         //指标停用
-        stopTag(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
+        stopTagCache(gasParametersTag.getAppId(), gasParametersTag.getPlatformCode(), gasParametersTag.getTagCode());
     }
 
     /**

cloud-consume/cloud-consume-biz/src/main/java/com/lutao/cloud/consume/factory/HazardousGasParametersTagStrategy.java

@@ -149,41 +149,53 @@ public class HazardousGasParametersTagStrategy implements CommandStrategy<Hazard
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void stopTag(String appId, String platformCode, String tagCode) {
+    private void stopTagCache(String appId, String platformCode, String tagCode) {
         log.info("hazardousGasParametersTagStrategy 进行stopTag操作 .......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
         BoundHashOperations parametersTagCache = hazardousGasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorHazardousGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
-
-        if (Objects.nonNull(count)) {
-            parametersTagCache.put("end_time", LocalDateTime.now().toString());
-            // 计算每分钟应该接受的数据量
-            LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-            Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
-            parametersTagCache.put("should_accept_count", shouldAcceptCount);
-            parametersTagCache.put("enable", 1);
 
+        // 2024/11/14 删除时候-判断缓存是否存在
+        Boolean hasKey = hazardousGasParametersTagStrategy.redisTemplate.hasKey(LocalDate.now() + ":" + PerceptionMonitorHazardousGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        if (Boolean.FALSE.equals(hasKey)){
+            return;
         }
+        parametersTagCache.put("end_time", LocalDateTime.now().toString());
+        // 计算每分钟应该接受的数据量
+        LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
+        Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
+        parametersTagCache.put("should_accept_count", shouldAcceptCount);
+        parametersTagCache.put("enable", 1);
     }
 
-    private void startTag(String appId, String platformCode, String tagCode) {
+    private void startTagCache(HazardousGasParametersTag hazardousGasParametersTag) {
         log.info("hazardousGasParametersTagStrategy 进行startTag操作 ......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
-        BoundHashOperations parametersTagCache = hazardousGasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorHazardousGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        BoundHashOperations parametersTagCache = hazardousGasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":"
+                + PerceptionMonitorHazardousGasInfo.TABLE_NAME + ":"
+                + hazardousGasParametersTag.getAppId() + ":"
+                + hazardousGasParametersTag.getPlatformCode() + ":"
+                + hazardousGasParametersTag.getTagCode());
         Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count) && count < 1) {
+        if (!Objects.nonNull(count)) {
+            // 初始化
             parametersTagCache.put("should_accept_count", 0);
             parametersTagCache.put("received_count", 0);
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.put("enable", 0);
-        }else {
-            //有的情况下
+            parametersTagCache.put("resource", PerceptionMonitorGasInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(hazardousGasParametersTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
+        } else {
+            // 恢复缓存
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.delete("end_time");
             parametersTagCache.put("enable", 0);
+            parametersTagCache.put("resource", PerceptionMonitorGasInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(hazardousGasParametersTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
         }
     }
 
@@ -194,19 +206,16 @@ public class HazardousGasParametersTagStrategy implements CommandStrategy<Hazard
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void addTag(String appId, String platformCode, String tagCode) {
+    private void addTagCache(String appId, String platformCode, String tagCode) {
         log.info("hazardousGasParametersTagStrategy 进行addTag操作 .......");
         BoundHashOperations parametersTagCache = hazardousGasParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorHazardousGasInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count)) {
-            parametersTagCache.put("begin_time", LocalDateTime.now().toString());
-            parametersTagCache.put("resource", PerceptionMonitorHazardousGasInfo.TABLE_NAME);
-            parametersTagCache.put("received_count", 0);
-            parametersTagCache.put("should_accept_count", 0);
-            parametersTagCache.put("enable", 0);
-            parametersTagCache.expire(48, TimeUnit.HOURS);
-        }
+        parametersTagCache.put("begin_time", LocalDateTime.now().toString());
+        parametersTagCache.put("resource", PerceptionMonitorHazardousGasInfo.TABLE_NAME);
+        parametersTagCache.put("received_count", 0);
+        parametersTagCache.put("should_accept_count", 0);
+        parametersTagCache.put("enable", 0);
+        parametersTagCache.expire(48, TimeUnit.HOURS);
     }
 
     /**
@@ -219,11 +228,9 @@ public class HazardousGasParametersTagStrategy implements CommandStrategy<Hazard
     private void add(HazardousGasParametersTag hazardousGasParametersTag) {
         log.info("fireParametersTagStrategy 进行add操作 ......");
         hazardousGasParametersTagStrategy.hazardousGasParametersTagService.save(hazardousGasParametersTag);
-
-        addTag(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
-        //指标停用
-        if ("1".equals(hazardousGasParametersTag.getStatus())) {
-            stopTag(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
+        // 新增启用指标操作缓存；新增停用的指标不操作缓存（指标为新增，数据库中不存在）
+        if (!"1".equals(hazardousGasParametersTag.getStatus())) {
+            addTagCache(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
         }
 
     }
@@ -246,13 +253,19 @@ public class HazardousGasParametersTagStrategy implements CommandStrategy<Hazard
                 .eq(HazardousGasParametersTag::getAppId, hazardousGasParametersTag.getAppId())
                 .eq(HazardousGasParametersTag::getOriginId, hazardousGasParametersTag.getOriginId()));
 
-        // 指标停用
+        /**
+         * 1.当前状态有效0，新指标状态有效0，不操作缓存；
+         * 2.当前状态有效0，新指标状态无效1，停止缓存；->stopTagCache()
+         * 3.当前状态无效1，新指标状态有效0，恢复缓存；->startTagCache()
+         * 4.当前状态无效1，新指标状态无效1，不操作缓存；
+         */
+        // 指标缓存停用
         if ("1".equals(hazardousGasParametersTag.getStatus()) && !"1".equals(hazardousGasParametersTagTemp.getStatus())) {
-            stopTag(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
+            stopTagCache(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
         }
-        // 启用
+        // 指标缓存启用
         if ("0".equals(hazardousGasParametersTag.getStatus()) && ("1".equals(hazardousGasParametersTagTemp.getStatus()))) {
-            startTag(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
+            startTagCache(hazardousGasParametersTag);
         }
     }
 
@@ -270,7 +283,7 @@ public class HazardousGasParametersTagStrategy implements CommandStrategy<Hazard
                 .eq(HazardousGasParametersTag::getOriginId, hazardousGasParametersTag.getOriginId()));
 
         //指标停用
-        stopTag(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
+        stopTagCache(hazardousGasParametersTag.getAppId(), hazardousGasParametersTag.getPlatformCode(), hazardousGasParametersTag.getTagCode());
     }
 
     /**

cloud-consume/cloud-consume-biz/src/main/java/com/lutao/cloud/consume/factory/ProcessParametersTagStrategy.java

@@ -127,7 +127,7 @@ public class ProcessParametersTagStrategy implements CommandStrategy<ProcessPara
         } else if (MessageCommandEnum.MESSAGE_DEL.getCommand().equals(command)) {
             log.info("processParametersTagStrategy 进行del-batch操作 .......删除数量为：" + processParameterTags.size());
             for (ProcessParameterTag processParameterTag : processParameterTags) {
-
+                del(processParameterTag);
             }
         } else if (MessageCommandEnum.MESSAGE_upd.getCommand().equals(command)) {
             log.info("processParametersTagStrategy 进行upd-batch操作 .......更新数量为：" + processParameterTags.size());
@@ -146,65 +146,74 @@ public class ProcessParametersTagStrategy implements CommandStrategy<ProcessPara
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void stopTag(String appId, String platformCode, String tagCode) {
+    private void stopTagCache(String appId, String platformCode, String tagCode) {
         log.info("processParametersTagStrategy 进行stopTag操作 .......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
         BoundHashOperations parametersTagCache = processParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
-
-        if (Objects.nonNull(count)) {
-            //这个指标信息已经存在 且收到信息的情况
-            parametersTagCache.put("end_time", LocalDateTime.now().toString());
-            // 计算每分钟应该接受的数据量
-            LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-            Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
-            parametersTagCache.put("should_accept_count", shouldAcceptCount);
-            parametersTagCache.put("enable", 1);
 
+        // 2024/11/14 删除时候-判断缓存是否存在
+        Boolean hasKey = processParametersTagStrategy.redisTemplate.hasKey(LocalDate.now() + ":" + PerceptionMonitorInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        if (Boolean.FALSE.equals(hasKey)){
+            return;
         }
+        //这个指标信息已经存在 且收到信息的情况
+        parametersTagCache.put("end_time", LocalDateTime.now().toString());
+        // 计算每分钟应该接受的数据量
+        LocalDateTime beginTime = LocalDateTime.parse(parametersTagCache.get("begin_time").toString(), DateTimeFormatter.ISO_LOCAL_DATE_TIME);
+        Long shouldAcceptCount = TimeUtil.getTimeDifference(beginTime, LocalDateTime.now());
+        parametersTagCache.put("should_accept_count", shouldAcceptCount);
+        parametersTagCache.put("enable", 1);
     }
 
-    private void startTag(String appId, String platformCode, String tagCode) {
+    private void startTagCache(ProcessParameterTag processParameterTag) {
         log.info("processParametersTagStrategy 进行startTag操作 ......");
         //停用 接口收到指标停用的变更后， 在线率停止计算，应收到次数以变更时间为节点，计算在线率
         //启用 接口收到指标启用的变更后，如果初始化中没有，则新增，开始时间以变更时间为准，计算在线率。 如果初始化中有（一天内多次停用启用，在线率分段计算，只计算启用段内的。
-        BoundHashOperations parametersTagCache = processParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
+        BoundHashOperations parametersTagCache = processParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":"
+                + PerceptionMonitorInfo.TABLE_NAME + ":"
+                + processParameterTag.getAppId() + ":"
+                + processParameterTag.getPlatformCode() + ":"
+                + processParameterTag.getTagCode());
         Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count) && count < 1) {
+        if (!Objects.nonNull(count)) {
+            // 初始化
             parametersTagCache.put("should_accept_count", 0);
             parametersTagCache.put("received_count", 0);
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.put("enable", 0);
-        }else{
-            //有的情况下
+            parametersTagCache.put("resource", PerceptionMonitorInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(processParameterTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
+        } else {
+            // 恢复缓存
             parametersTagCache.put("begin_time", LocalDateTime.now().toString());
             parametersTagCache.delete("end_time");
             parametersTagCache.put("enable", 0);
+            parametersTagCache.put("resource", PerceptionMonitorInfo.TABLE_NAME);
+            parametersTagCache.put("tagInfo", JSONUtil.toJsonStr(processParameterTag));
+            parametersTagCache.expire(48, TimeUnit.HOURS);
         }
     }
 
     /**
-     * @Description: 指标添加
+     * @Description: 指标缓存添加
      * @Param: [java.lang.String, java.lang.String, java.lang.String]
      * @Return: void
      * @Author: vincent.zhao
      * @Date: 2024/6/20 16:01
      */
-    private void addTag(String appId, String platformCode, String tagCode) {
+    private void addTagCache(String appId, String platformCode, String tagCode) {
         log.info("processParametersTagStrategy 进行addTag操作 ......");
         BoundHashOperations parametersTagCache = processParametersTagStrategy.redisTemplate.boundHashOps(LocalDate.now() + ":" + PerceptionMonitorInfo.TABLE_NAME + ":" + appId + ":" + platformCode + ":" + tagCode);
-        Integer count = (Integer) parametersTagCache.get("received_count");
 
-        if (Objects.nonNull(count)) {
-            parametersTagCache.put("begin_time", LocalDateTime.now().toString());
-            parametersTagCache.put("resource", PerceptionMonitorFireInfo.TABLE_NAME);
-            parametersTagCache.put("received_count", 0);
-            parametersTagCache.put("should_accept_count", 0);
-            parametersTagCache.put("enable", 0);
-            parametersTagCache.expire(48, TimeUnit.HOURS);
-        }
+        parametersTagCache.put("begin_time", LocalDateTime.now().toString());
+        parametersTagCache.put("resource", PerceptionMonitorInfo.TABLE_NAME);
+        parametersTagCache.put("received_count", 0);
+        parametersTagCache.put("should_accept_count", 0);
+        parametersTagCache.put("enable", 0);
+        parametersTagCache.expire(48, TimeUnit.HOURS);
     }
 
     /**
@@ -217,10 +226,9 @@ public class ProcessParametersTagStrategy implements CommandStrategy<ProcessPara
     private void add(ProcessParameterTag processParameterTag) {
         log.info("processParametersTagStrategy 进行add操作 ......");
         processParametersTagStrategy.processParameterTagService.save(processParameterTag);
-        addTag(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
-        //指标停用
-        if ("1".equals(processParameterTag.getStatus())) {
-            stopTag(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
+        // 新增启用指标操作缓存；新增停用的指标不操作缓存（指标为新增，数据库中不存在）
+        if (!"1".equals(processParameterTag.getStatus())) {
+            addTagCache(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
         }
     }
 
@@ -240,13 +248,20 @@ public class ProcessParametersTagStrategy implements CommandStrategy<ProcessPara
         processParametersTagStrategy.processParameterTagService.update(processParameterTag, Wrappers.lambdaQuery(ProcessParameterTag.class)
                 .eq(ProcessParameterTag::getAppId, processParameterTag.getAppId())
                 .eq(ProcessParameterTag::getOriginId, processParameterTag.getOriginId()));
-        // 指标停用
+
+        /**
+         * 1.当前状态有效0，新指标状态有效0，不操作缓存；
+         * 2.当前状态有效0，新指标状态无效1，停止缓存；->stopTagCache()
+         * 3.当前状态无效1，新指标状态有效0，恢复缓存；->startTagCache()
+         * 4.当前状态无效1，新指标状态无效1，不操作缓存；
+         */
+        // 指标缓存停用
         if ("1".equals(processParameterTag.getStatus()) && !"1".equals(processParametersTagTemp.getStatus())) {
-            stopTag(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
+            stopTagCache(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
         }
-        // 启用
+        // 指标缓存启用
         if ("0".equals(processParameterTag.getStatus()) && ("1".equals(processParametersTagTemp.getStatus()))) {
-            startTag(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
+            startTagCache(processParameterTag);
         }
 
     }
@@ -265,10 +280,17 @@ public class ProcessParametersTagStrategy implements CommandStrategy<ProcessPara
                 .eq(ProcessParameterTag::getAppId, processParameterTag.getAppId())
                 .eq(ProcessParameterTag::getOriginId, processParameterTag.getOriginId()));
 
-        //指标停用
-        stopTag(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
+        // 指标缓存停用
+        stopTagCache(processParameterTag.getAppId(), processParameterTag.getPlatformCode(), processParameterTag.getTagCode());
     }
 
+    /**
+     * @Description: 清理重复数据
+     * @Param: [com.lutao.entity.ProcessParameterTag]
+     * @Return: void
+     * @Author: vincent.zhao
+     * @Date: 2024/11/12 19:30
+     */
     public void checkData4Add(ProcessParameterTag processParameterTag) {
         int count = processParametersTagStrategy.processParameterTagService.count(Wrappers.lambdaQuery(ProcessParameterTag.class).eq(ProcessParameterTag::getAppId, processParameterTag.getAppId())
                 .eq(ProcessParameterTag::getTagCode, processParameterTag.getTagCode()));

cloud-oil/cloud-oil-biz/src/main/java/com/lutao/cloud/oil/typhoon/BusiConstant.java

@@ -15,74 +15,23 @@ public class BusiConstant {
     // 应急部资源池信息 start
     /**
      * 应急部资源池信息
-     * 应用-预警中心
      * 应用代码-01
      */
-    public static String APP_USER_NAME_01 = "ad1d7c9694ed4258ad3ef00b7599bdc1"; // client_id
-    public static String APP_PASSWORD_01 = "26f2d51f3c5a4253aff75302c0d8a844"; // client_secret
-
-    // 对比订阅
-    public static String YJZX_DZ_TASK_ID = "YJZX-DZ-DEV013"; // 地震对比订阅id；唯一标识
-    public static String YJZX_DZ_RULE_ID = "YJZX-DZ-DEV013-R"; // 地震对比订阅规则id；唯一标识
-
-    /**
-     * 应急部资源池信息
-     * 应用-互联网+执法系统
-     * 应用代码-02
-     */
-//    public static String APP_USER_NAME_02 = "b26bc16dbf784d9f992b6fc9baddd2f4"; // client_id
-//    public static String APP_USER_NAME_02 = "dac02cefbb1b407e9cf3420cd0303880"; // client_secret
+    public static String APP_USER_NAME_01 = "8a7b785f08d04d439b71bc2b415cb685"; // client_id
+    public static String APP_PASSWORD_01 = "OA!Bq_P!"; // client_secret
 
     /**
      * 应用的用户名密码常量
      */
     public static Map<String, String> APP_USER_NAME_PASSWORD = new HashMap() {{
         put(APP_USER_NAME_01, APP_PASSWORD_01);
-//        put(APP_USER_NAME_02, APP_USER_NAME_02);
     }};
 
-    /**
-     * 通用查询V1接口
-     */
-    public static String QUERY_INFO_URL = "/service/api/general/data/queryInfo";
-    public static String QUERY_INFO_V2_URL = "/service/api/general/data/queryInfo?version=v2";
-    /**
-     * 地震速报系统_地震速报实时信息（数据服务_获取比对订阅数据）
-     */
-    public static String COMPARE_SUBSCRIBE_DATA = "/service/api/general/subscription/data/compareSubscribeData";
-
     public static String IP_59_255_61_5 = "10.18.17.84"; // ip 59_255_61_5
-    public static String IP_59_255_61_6 = "10.18.17.85"; // ip 59_255_61_6
     public static String PORT_9090 = "9090"; // 端口
 
     public static String TAKEN_URL = "/oauth/token"; // token URL
-
-    public static String DATA_SEND_FLAG = "110000000001"; // 数据发送方标识
-    public static String DATA_RECEIVE_FLAG = "110000000001"; // 数据接收方标识
-
-    /**
-     * 资源池登录后token
-     */
-    public static volatile String YJBZYC_TOKEN_59_255_61_5 = "";
-    public static volatile String YJBZYC_TOKEN_59_255_61_6 = "";
-    public static volatile String YJBZYC_TOKEN_10 = "";
     // 应急部资源池信息 end
 
 
-    /**
-     * 省代码和简称
-     */
-    public static Map<String, String> ARCODEMAP = new HashMap<>();
-
-    /**
-     * 强降雨企业列表
-     */
-    public volatile static List<String> QIANG_JIANG_YU_COM_LIST = new ArrayList<>();
-
-    /**
-     * 是否生产环境标识
-     */
-    public static Boolean PRODUCTION_ENVIRONMENT = false;
-
-
 }

cloud-oil/cloud-oil-biz/src/main/java/com/lutao/cloud/oil/typhoon/DataAPIUtils.java

@@ -17,11 +17,11 @@ import java.io.InputStream;
 import java.io.InputStreamReader;
 
 /**
- * @author zhaoqiang
- * @title: DataAPIUtils.java
- * @package: org.casst.back.task.util
- * @description: 数据接口工具类
- * @created 2020年11月4日 上午2:16:39
+ * @Description: 数据接口工具类
+ * @Param:
+ * @Return:
+ * @Author: vincent.zhao
+ * @Date: 2024/11/18 18:36
  */
 @RefreshScope
 public class DataAPIUtils {

cloud-oil/cloud-oil-biz/src/main/resources/bootstrap.yml

@@ -1,13 +1,15 @@
 server:
-  port: 9299
+  port: 9300
 spring:
   application:
     name: @artifactId@
   cloud:
     nacos:
       discovery:
-        ip: ${hostIp}
-        server-addr: ${cloudRegister}
+        #server-addr: ${NACOS_HOST:172.16.2.150}:${NACOS_PORT:58840}
+        server-addr: ${NACOS_HOST:localhost}:${NACOS_PORT:8848}
+        #server-addr: ${NACOS_HOST:192.168.31.68}:${NACOS_PORT:8848}
+        #ip: ${ip}
       config:
         server-addr: ${spring.cloud.nacos.discovery.server-addr}
         file-extension: yml