Переглянути джерело

增加处理数据中台增量同步报警条件的逻辑

lixing 3 роки тому
батько
коміт
611dad9123

+ 26 - 0
AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/AlarmEngineMsgHandler.java

@@ -9,6 +9,9 @@ import io.netty.channel.ChannelHandler;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.util.CollectionUtils;
+
+import java.util.Map;
 
 /**
  * @description: 报警引擎消息处理类
@@ -145,4 +148,27 @@ public class AlarmEngineMsgHandler {
         // 先删除,再创建
         // TODO: 2021/11/16 更新redis
     }
+
+    /**
+     * 增量同步报警定义
+     *
+     * @author lixing
+     * @version V1.0 2021/11/20 3:08 下午
+     */
+    @SneakyThrows
+    public void incrementSyncAlarmConfig(DmpMessage dmpMessage) {
+        Map<String, JSONArray> changedAlarmConfigs = alarmRecordMsgHandler.queryChangedAlarmConfigs(dmpMessage);
+        JSONArray createdConfigUniques = changedAlarmConfigs.get("createdConfigUniques");
+        JSONArray deletedConfigUniques = changedAlarmConfigs.get("deletedConfigUniques");
+        if (!CollectionUtils.isEmpty(deletedConfigUniques)) {
+            // 处理删除的报警定义 // TODO: 2021/11/20
+            // 报警定义中不包含报警条件的id
+            // 需要根据报警定义中的报警类型,从设备与报警条件的关联关系中找到对应的报警条件id。
+            // 删除报警条件
+            // 删除关联关系
+        }
+        if (!CollectionUtils.isEmpty(createdConfigUniques)) {
+            // 处理新增的报警定义 // TODO: 2021/11/20
+        }
+    }
 }

+ 70 - 6
AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/JmsConfig.java

@@ -60,6 +60,11 @@ public class JmsConfig {
     private final String objConditionRelRoutingKey = "obj-condition-rel-routing-key";
 
     /**
+     * 报警定义路由键
+     */
+    private static final String alarmConfigRoutingKey = "alarm-config-routing-key";
+
+    /**
      * 报警记录 路由键
      */
     private final String alarmRecordRoutingKey = "alarm-msg-routing-key";
@@ -70,6 +75,11 @@ public class JmsConfig {
     private final String conditionQueue = "condition-queue";
 
     /**
+     * 报警定义队列(数据中台设备的报警条件)
+     */
+    private final String alarmConfigQueue = "alarm-config-queue";
+
+    /**
      * 设备与报警条件关联关系队列
      */
     private final String objConditionRelQueue = "obj-condition-rel-queue";
@@ -89,6 +99,11 @@ public class JmsConfig {
     }
 
     @Bean
+    public Queue alarmConfigQueue() {
+        return new Queue(alarmConfigQueue, true);
+    }
+
+    @Bean
     public Queue alarmRecordQueue() {
         return new Queue(alarmRecordQueue, true);
     }
@@ -109,13 +124,24 @@ public class JmsConfig {
     }
 
     @Bean
+    public Binding alarmConfigBinding() {
+        return BindingBuilder.bind(alarmConfigQueue()).to(alarmServiceExchange()).with(alarmConfigRoutingKey);
+    }
+
+    @Bean
     public Binding alarmRecordBinding() {
         return BindingBuilder.bind(alarmRecordQueue()).to(alarmServiceExchange()).with(alarmRecordRoutingKey);
     }
 
+    /**
+     * 处理业务中台的消息
+     *
+     * @author lixing
+     * @version V1.0 2021/11/19 10:43 下午
+     */
     @RabbitHandler
     @RabbitListener(queues = {conditionQueue, objConditionRelQueue})
-    public void dealMsg(DmpMessage msg, Channel channel, Message message) throws Exception {
+    public void dealAlarmServiceMsg(DmpMessage msg, Channel channel, Message message) throws Exception {
         try {
             log.debug("============================== Receive:" + msg);
             if (JmsTypeEnum.NEW_CONDITION.getType().equals(msg.getType())) {
@@ -142,12 +168,21 @@ public class JmsConfig {
         channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
     }
 
+    /**
+     * 处理报警引擎报警消息
+     *
+     * @author lixing
+     * @version V1.0 2021/11/20 2:19 下午
+     */
     @RabbitHandler
     @RabbitListener(queues = {alarmRecordQueue})
     public void dealAlarmRecordMsg(DmpMessage msg, Channel channel, Message message) throws Exception {
         try {
             // 根据报警消息获取报警系统
             String alarmCategory = msgHandler.getAlarmCategory(msg);
+            if (StringUtils.isBlank(alarmCategory)) {
+                return;
+            }
             HashSet<String> categorySet = Sets.newHashSet(categoryList);
             if (!categorySet.contains(alarmCategory)) {
                 // 如果子系统不能处理此报警系统的消息,将消息扔回消息队列
@@ -174,15 +209,44 @@ public class JmsConfig {
         channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
     }
 
-
+    /**
+     * redis解锁
+     *
+     * @param key 要解锁的key
+     * @author lixing
+     * @version V1.0 2021/11/20 2:19 下午
+     */
     public void unlock(String key) {
         try {
-            String currentValue = String.valueOf(redisTemplate.opsForValue().get(key));
-            if (!StringUtils.isEmpty(currentValue)) {
-                redisTemplate.opsForValue().getOperations().delete(key);
-            }
+            redisTemplate.delete(key);
         } catch (Exception e) {
             e.printStackTrace();
         }
     }
+
+    /**
+     * 数据中台消息处理
+     *
+     * @author lixing
+     * @version V1.0 2021/11/20 2:20 下午
+     */
+    @RabbitHandler
+    @RabbitListener(queues = {alarmConfigQueue})
+    public void dealDmpAlarmMsg(DmpMessage msg, Channel channel, Message message) {
+        try {
+            log.debug("============================== Receive:" + msg);
+            //报警定义变化
+            if (JmsTypeEnum.ALARM_CONFIGS_CHANGE.getType().equals(msg.getType())) {
+                log.debug("================收到一条报警定义变化通知==============");
+                try {
+                    msgHandler.incrementSyncAlarmConfig(msg);
+                } catch (Exception e) {
+                    log.error("error", e);
+                }
+            }
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
+        } catch (Exception e) {
+            log.error("消息消费失败,{}", e.getMessage());
+        }
+    }
 }

+ 2 - 1
AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/JmsTypeEnum.java

@@ -23,7 +23,8 @@ public enum JmsTypeEnum {
     DELETE_CONDITION("delete_condition", "删除报警条件"),
     NEW_OBJ_CONDITION_REL("new_obj_condition_rel", "新增条件和设备的关联关系"),
     DELETE_OBJ_CONDITION_REL("delete_obj_condition_rel", "删除条件和设备的关联关系"),
-    UPDATE_OBJ_CONDITION_REL("update_obj_condition_rel", "更新条件和设备的关联关系");
+    UPDATE_OBJ_CONDITION_REL("update_obj_condition_rel", "更新条件和设备的关联关系"),
+    ALARM_CONFIGS_CHANGE("alarm_configs_change", "数据中台报警条件更新");
 
 
     @Setter