向报警引擎同步报警条件和关联关系

AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/feign/fallback/AlarmClientFallbackFactory.java

@@ -3,6 +3,7 @@ package com.persagy.apm.energyalarmstarter.alarmdata.feign.fallback;
 import com.persagy.apm.energyalarmstarter.alarmdata.feign.client.AlarmClient;
 import feign.hystrix.FallbackFactory;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
 
 /**
  * 降级处理
@@ -11,6 +12,7 @@ import lombok.extern.slf4j.Slf4j;
  * @version V1.0 2021/9/15 10:59 下午
  */
 @Slf4j
+@Component
 public class AlarmClientFallbackFactory implements FallbackFactory<AlarmClient> {
     @Override
     public AlarmClient create(Throwable throwable) {

AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/model/dto/AlarmCondition.java

@@ -1,5 +1,6 @@
 package com.persagy.apm.energyalarmstarter.alarmdata.model.dto;
 
+import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import lombok.Data;
@@ -38,9 +39,13 @@ public class AlarmCondition {
     /**
      * 存储用户自定义的配置
      */
-    private JSONObject configs;
+    private JSONArray configs;
     /**
      * 报警规则
      */
     private List<String> rules;
+    /**
+     * 报警生效时间
+     */
+    private AlarmEffectTime effectTime;
 }

AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/model/dto/AlarmEffectTime.java

@@ -0,0 +1,19 @@
+package com.persagy.apm.energyalarmstarter.alarmdata.model.dto;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import lombok.Data;
+
+/**
+ * 报警生效时间
+ *
+ * @author lixing
+ * @version V1.0 2021/9/15 10:32 下午
+ **/
+@Data
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class AlarmEffectTime {
+    /**
+     * 生效时间类型，period/always
+     */
+    private String type;
+}

AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/model/dto/alarmrecord/BatchUpdateAlarmRecordDTO.java

@@ -17,5 +17,5 @@ import java.util.List;
 @Data
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public class BatchUpdateAlarmRecordDTO extends UrlParamDTO {
-    private List<UpdateAlarmRecordDTO> alarmRecords;
+    private List<UpdateAlarmRecordDTO> alarmRecordModels;
 }

AlarmDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmdata/service/AlarmConfigServiceImpl.java

@@ -110,7 +110,7 @@ public class AlarmConfigServiceImpl {
      */
     public DmpResult<DmpBatchUpsertVO> batchUpdate(BatchUpdateAlarmConfigDTO batchUpdateAlarmConfigDTO) throws Exception {
         JSONObject jsonObject = Obj2JSONObject.convert(batchUpdateAlarmConfigDTO);
-        DmpResult<JSONObject> result = alarmClient.batchCreateAlarmConfig(batchUpdateAlarmConfigDTO.fetchParams(), jsonObject);
+        DmpResult<JSONObject> result = alarmClient.batchUpdateAlarmConfig(batchUpdateAlarmConfigDTO.fetchParams(), jsonObject);
         return DmpResultUtil.singleConvert(result, DmpBatchUpsertVO.class);
     }
 

AlarmEngineStarter/pom.xml

@@ -81,5 +81,12 @@
             <version>v1.0.0</version>
         </dependency>
 
+        <!-- apm-common -->
+        <dependency>
+            <groupId>com.persagy</groupId>
+            <artifactId>apm-common</artifactId>
+            <version>v1.0.0</version>
+        </dependency>
+
     </dependencies>
 </project>

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/AlarmCondition.java

@@ -0,0 +1,80 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign;
+
+import com.alibaba.fastjson.JSONObject;
+import lombok.Builder;
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * 报警条件
+ *
+ * @author lixing
+ * @version V1.0 2021/10/22 9:42 上午
+ **/
+@Data
+@Builder
+public class AlarmCondition {
+    /**
+     * 主键
+     */
+    private String id;
+    
+    /**
+     * 包含的参数
+     */
+    private List<String> infoCodes;
+
+    /**
+     * 触发条件-前端展示
+     */
+    private List<List<CalculateMethodDTO>> triggerFrontend;
+
+    /**
+     * 触发条件-后端使用
+     */
+    private String triggerBackend;
+
+    /**
+     * 触发时长
+     */
+    private Integer triggerUphold;
+
+    /**
+     *报警等级
+     */
+    private String alarmLevel;
+
+    /**
+     * 持续多久后恢复
+     */
+    private Integer endUphold;
+
+    /**
+     * 触发规则-中文描述
+     */
+    private List<String> triggerRules;
+
+//    @ApiModelProperty("恢复条件-前端")
+//    private List<List<CalculateMethodDTO>> endFrontend;
+
+    /**
+     * 恢复条件-后端
+     */
+    private String endBackend;
+
+    /**
+     * 恢复规则-中文描述
+     */
+    private List<String> endRules;
+
+    /**
+     * 生效时间类型
+     */
+    private String effectTimeType;
+
+    /**
+     * 生效时间详情
+     */
+    private JSONObject effectTimeDetail;
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/BaseRequestBody.java

@@ -0,0 +1,27 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign;
+
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * 基础请求类
+ *
+ * @author lixing
+ * @version V1.0 2021/10/22 2:23 下午
+ **/
+@Data
+@Builder
+public class BaseRequestBody {
+    /**
+     * 当前用户id
+     */
+    private String userId;
+    /**
+     * 集团编码
+     */
+    private String groupCode;
+    /**
+     * 项目id
+     */
+    private String projectId;
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/CalculateMethodDTO.java

@@ -0,0 +1,40 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign;
+
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * 报警规则
+ *
+ * @author lixing
+ * @version V1.0 2021-09-08 22:43:58
+ */
+@Data
+public class CalculateMethodDTO {
+    /**
+     * 操作符
+     */
+    private String operator;
+    
+    /**
+     * 公式
+     */
+    private List<Formula> formula;
+    
+    /**
+     * 监测指标id
+     */
+    private String monitorIndicatorId;
+
+    /**
+     * 监测指标name
+     */
+    private String monitorIndicatorName;
+    
+    /**
+     * 临时存储公式
+     */
+    private String tempFormula;
+    
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/Formula.java

@@ -0,0 +1,21 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 公式实体类
+ *
+ * @author lixing
+ * @version V1.0 2021-09-08 22:30:38
+ */
+@Data
+public class Formula implements Serializable {
+    private static final long serialVersionUID = 268657175929005281L;
+
+    private String key;
+    private String label;
+    private String value;
+
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/ObjConditionRel.java

@@ -0,0 +1,31 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign;
+
+import lombok.Builder;
+import lombok.Data;
+
+/**
+ * 设备与报警条件的关联
+ *
+ * @author lixing
+ * @version V1.0 2021/10/22 9:42 上午
+ **/
+@Data
+@Builder
+public class ObjConditionRel {
+    /**
+     * 报警对象id
+     */
+    private String objId;
+    /**
+     * 报警类型编码
+     */
+    private String itemCode;
+    /**
+     * 项目id
+     */
+    private String projectId;
+    /**
+     * 报警条件id
+     */
+    private String conditionId;
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/ProjectVO.java

@@ -0,0 +1,28 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign;
+
+import lombok.Data;
+
+/**
+ * 项目vo类
+ *
+ * @author lixing
+ * @version V1.0 2021/10/22 2:19 下午
+ */
+@Data
+public class ProjectVO {
+
+    /**
+     * 集团编码
+     */
+    private String groupCode;
+
+    /**
+     * 项目名称
+     */
+    private String projectName;
+
+    /**
+     * 项目id
+     */
+    private String projectId;
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/client/EnergyAlarmServiceClient.java

@@ -0,0 +1,53 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign.client;
+
+import com.persagy.apm.common.response.CommonResult;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.*;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.fallback.EnergyAlarmServiceClientFallbackFactory;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+
+import java.util.List;
+
+/**
+ * @description: feign调用数据中台报警接口
+ * @author: lixing
+ * @company: Persagy Technology Co.,Ltd
+ * @since: 2020/11/27 10:43 上午
+ * @version: V1.0
+ */
+@FeignClient(name = "energy-alarm-service", fallbackFactory = EnergyAlarmServiceClientFallbackFactory.class)
+public interface EnergyAlarmServiceClient {
+    /**
+     * 全量查询报警条件
+     *
+     * @param baseRequestBody 查询条件
+     * @return 报警条件列表
+     * @author lixing
+     * @version V1.0 2021/10/22 10:33 上午
+     */
+    @PostMapping("alarmConditions/query_all")
+    CommonResult<List<AlarmCondition>> queryAllAlarmCondition(@RequestBody BaseRequestBody baseRequestBody);
+
+    /**
+     * 全量查询项目
+     *
+     * @param baseRequestBody 查询条件
+     * @return 项目列表
+     * @author lixing
+     * @version V1.0 2021/10/22 2:33 下午
+     */
+    @PostMapping("projectAlarmRules/projects")
+    CommonResult<List<ProjectVO>> queryProjects(@RequestBody BaseRequestBody baseRequestBody);
+
+    /**
+     * 全量查询项目上设备与报警条件的关联
+     *
+     * @param baseRequestBody 查询条件
+     * @return 设备与报警条件的关联
+     * @author lixing
+     * @version V1.0 2021/10/22 2:33 下午
+     */
+    @PostMapping("/projectAlarmRuleObjRels/query_obj_condition_rels")
+    CommonResult<List<ObjConditionRel>> queryObjAlarmConditionRel(@RequestBody BaseRequestBody baseRequestBody);
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/fallback/EnergyAlarmServiceClientFallbackFactory.java

@@ -0,0 +1,21 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign.fallback;
+
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.client.EnergyAlarmServiceClient;
+import feign.hystrix.FallbackFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * 降级处理
+ *
+ * @author lixing
+ * @version V1.0 2021/9/15 10:59 下午
+ */
+@Slf4j
+@Component
+public class EnergyAlarmServiceClientFallbackFactory implements FallbackFactory<EnergyAlarmServiceClient> {
+    @Override
+    public EnergyAlarmServiceClient create(Throwable throwable) {
+       return null;
+    }
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/feign/service/AlarmServiceImpl.java

@@ -0,0 +1,80 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.feign.service;
+
+import com.persagy.apm.common.response.CommonResult;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.AlarmCondition;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.BaseRequestBody;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.ObjConditionRel;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.ProjectVO;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.client.EnergyAlarmServiceClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 报警业务层service类
+ *
+ * @author lixing
+ * @version V1.0 2021/10/22 2:36 下午
+ **/
+@Service
+public class AlarmServiceImpl {
+    @Autowired
+    EnergyAlarmServiceClient energyAlarmServiceClient;
+
+    /**
+     * 全量查询报警条件
+     *
+     * @param userId    用户id
+     * @param groupCode 集团编码
+     * @return 报警条件列表
+     * @author lixing
+     * @version V1.0 2021/10/22 10:33 上午
+     */
+    public List<AlarmCondition> queryAllAlarmCondition(String userId, String groupCode) {
+        CommonResult<List<AlarmCondition>> listCommonResult = energyAlarmServiceClient.queryAllAlarmCondition(
+                BaseRequestBody.builder().userId(userId).groupCode(groupCode).build());
+        if (listCommonResult == null) {
+            return new ArrayList<>();
+        }
+        return listCommonResult.getContent();
+    }
+
+    /**
+     * 全量查询项目
+     *
+     * @param userId    用户id
+     * @param groupCode 集团编码
+     * @return 项目列表
+     * @author lixing
+     * @version V1.0 2021/10/22 2:33 下午
+     */
+    public List<ProjectVO> queryProjects(String userId, String groupCode) {
+        CommonResult<List<ProjectVO>> listCommonResult = energyAlarmServiceClient.queryProjects(
+                BaseRequestBody.builder().userId(userId).groupCode(groupCode).build());
+        if (listCommonResult == null) {
+            return new ArrayList<>();
+        }
+        return listCommonResult.getContent();
+    }
+
+    /**
+     * 全量查询项目上设备与报警条件的关联
+     *
+     * @param userId    用户id
+     * @param groupCode 集团编码
+     * @param projectId 项目id
+     * @return 设备与报警条件的关联
+     * @author lixing
+     * @version V1.0 2021/10/22 2:33 下午
+     */
+    public List<ObjConditionRel> queryObjAlarmConditionRel(String userId, String groupCode, String projectId) {
+        CommonResult<List<ObjConditionRel>> listCommonResult = energyAlarmServiceClient.queryObjAlarmConditionRel(
+                BaseRequestBody.builder().userId(userId).groupCode(groupCode).projectId(projectId).build());
+        if (listCommonResult == null) {
+            return new ArrayList<>();
+        }
+        return listCommonResult.getContent();
+    }
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/JmsConfig.java

@@ -2,14 +2,12 @@ package com.persagy.apm.energyalarmstarter.alarmengine.jms;
 
 import com.persagy.apm.energyalarmstarter.alarmengine.jms.model.DmpMessage;
 import com.persagy.apm.energyalarmstarter.alarmengine.netty.NettyAlarmMsgBaseHandler;
-import com.persagy.apm.energyalarmstarter.alarmengine.util.StringUtil;
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.core.*;
 import org.springframework.amqp.rabbit.annotation.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -29,60 +27,80 @@ public class JmsConfig {
      */
     @Autowired
     private NettyAlarmMsgBaseHandler msgHandler;
+
     /**
-     * 报警定义变化类型
+     * 交换机
      */
-    private static final String ALARM_CONFIGS_CHANGE = "alarmConfigsChange";
+    private final String exchange = "exchange-alarm-service";
+
     /**
-     * 报警交换器
+     * 报警条件路由键
      */
-    @Value("${dmp.alarm.exchange}")
-    private String dmpAlarmExchange;
+    private final String alarmConditionRoutingKey = "alarm-condition-routing-key";
+
     /**
-     * 报警路由键
+     * 报警对象与报警条件关联关系路由键
      */
-    @Value("${dmp.alarm.routingKey}")
-    private String alarmRoutingKey;
+    private final String objConditionRelRoutingKey = "obj-condition-rel-routing-key";
+
     /**
-     * 报警队列
+     * 报警条件队列
      */
-    @Value("${dmp.alarm.queue}")
-    private String alarmQueue;
+    private final String conditionQueue = "condition-queue";
+
+    /**
+     * 设备与报警条件关联关系队列
+     */
+    private final String objConditionRelQueue = "obj-condition-rel-queue";
 
     @Bean
-    public Queue alarmQueue() {
-        return new Queue(alarmQueue, true);
+    public Queue conditionQueue() {
+        return new Queue(conditionQueue, true);
     }
 
     @Bean
-    public TopicExchange alarmExchange() {
-        return new TopicExchange(dmpAlarmExchange);
+    public Queue objConditionRelQueue() {
+        return new Queue(objConditionRelQueue, true);
     }
 
     @Bean
-    public Binding alarmBinding() {
-        return BindingBuilder.bind(alarmQueue()).to(alarmExchange()).with(alarmRoutingKey);
+    public TopicExchange alarmServiceExchange() {
+        return new TopicExchange(exchange);
+    }
+
+    @Bean
+    public Binding conditionBinding() {
+        return BindingBuilder.bind(conditionQueue()).to(alarmServiceExchange()).with(alarmConditionRoutingKey);
+    }
+
+    @Bean
+    public Binding objConditionRelBinding() {
+        return BindingBuilder.bind(objConditionRelQueue()).to(alarmServiceExchange()).with(objConditionRelRoutingKey);
     }
 
     @RabbitHandler
-    @RabbitListener(queues = "${dmp.alarm.queue}")
-    public void planQueues(String msg, Channel channel, Message message) {
+    @RabbitListener(queues = {conditionQueue, objConditionRelQueue})
+    public void dealMsg(DmpMessage msg, Channel channel, Message message) throws Exception {
         try {
-            log.info("============================== Receive:" + msg);
-            DmpMessage dmpMessage = StringUtil.transferItemToDTO(msg, DmpMessage.class);
-            //报警定义变化
-            if (ALARM_CONFIGS_CHANGE.equals(dmpMessage.getType())) {
-                log.info("================收到一条报警定义变化通知==============");
-                log.info(msg);
-                try {
-                    msgHandler.incrementSyncAlarmConfig(dmpMessage);
-                } catch (Exception e) {
-                    log.error("error", e);
-                }
+            log.debug("============================== Receive:" + msg);
+            if (JmsTypeEnum.NEW_CONDITION.getType().equals(msg.getType())) {
+                msgHandler.syncNewCondition(msg);
+            }
+            if (JmsTypeEnum.UPDATE_CONDITION.getType().equals(msg.getType())) {
+                msgHandler.syncUpdatedCondition(msg);
+            }
+            if (JmsTypeEnum.DELETE_CONDITION.getType().equals(msg.getType())) {
+                msgHandler.syncDeletedCondition(msg);
+            }
+            if (JmsTypeEnum.NEW_OBJ_CONDITION_REL.getType().equals(msg.getType())) {
+                msgHandler.syncNewObjConditionRelList(msg);
+            }
+            if (JmsTypeEnum.DELETE_OBJ_CONDITION_REL.getType().equals(msg.getType())) {
+                msgHandler.syncDeletedObjConditionRelList(msg);
             }
-            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
         } catch (Exception e) {
             log.error("消息消费失败，{}", e.getMessage());
         }
+        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
     }
 }

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/JmsTypeEnum.java

@@ -0,0 +1,57 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.jms;
+
+import com.persagy.apm.common.model.annotation.SwaggerDisplayEnum;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.commons.lang.StringUtils;
+
+/**
+ * 消息类型
+ *
+ * @author lixing
+ * @version V1.0 2021/9/13 10:01 上午
+ **/
+@SwaggerDisplayEnum
+@AllArgsConstructor
+public enum JmsTypeEnum {
+    /**
+     * 字典类型
+     */
+    NEW_CONDITION("new_condition", "新增报警条件"),
+    UPDATE_CONDITION("update_condition", "更新报警条件"),
+    DELETE_CONDITION("delete_condition", "删除报警条件"),
+    NEW_OBJ_CONDITION_REL("new_obj_condition_rel", "新增条件和设备的关联关系"),
+    DELETE_OBJ_CONDITION_REL("delete_obj_condition_rel", "删除条件和设备的关联关系");
+
+
+    @Setter
+    @Getter
+    private String type;
+    @Setter
+    @Getter
+    private String desc;
+
+    /**
+     * 根据type获取到枚举对象
+     *
+     * @param type 报告状态编码
+     * @return 报告状态枚举对象
+     * @author lixing
+     * @version V1.0 2021/5/31 4:06 下午
+     */
+    public static JmsTypeEnum getByType(String type) {
+        if (StringUtils.isBlank(type)) {
+            return null;
+        }
+        JmsTypeEnum[] enums = JmsTypeEnum.values();
+        for (JmsTypeEnum item : enums) {
+            if (type.equals(item.getType())) {
+                return item;
+            }
+        }
+        return null;
+    }
+}
+
+

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/netty/NettyAlarmMessage.java

@@ -27,20 +27,8 @@ public class NettyAlarmMessage<T> {
     @JSONField()
     private int version = 1;
 
-    /**
-     * 操作类型：
-     * 200-建立连接，200的source是projectId
-     * 1-请求、2 -响应、3-通知、
-     * 4-边缘端获取报警定义、
-     * 5-边缘端主动推送报警记录、
-     * 6-边缘端主动更新报警记录状态、
-     * 7-云端推送修改的报警定义给边缘端(增量新增修改报警定义)、
-     * 8-云端把报警记录的id推送到边缘端
-     * 9-边缘端取报警定义，云端推送给边缘端的标记(全量报警定义)
-     * 10-云端推送删除的报警定义给边缘端(增量删除报警定义)、
-     */
     @JSONField()
-    private int opCode;
+    private NettyMsgTypeEnum opCode;
 
     /**
      * 请求来源
@@ -71,7 +59,7 @@ public class NettyAlarmMessage<T> {
         return JSONObject.toJSONString(this);
     }
 
-    public NettyAlarmMessage(int opCode, List<T> content) {
+    public NettyAlarmMessage(NettyMsgTypeEnum opCode, List<T> content) {
         this.opCode = opCode;
         this.content = content;
     }

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/netty/NettyAlarmMsgBaseHandler.java

@@ -2,6 +2,11 @@ package com.persagy.apm.energyalarmstarter.alarmengine.netty;
 
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.Lists;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.AlarmCondition;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.ObjConditionRel;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.ProjectVO;
+import com.persagy.apm.energyalarmstarter.alarmengine.feign.service.AlarmServiceImpl;
 import com.persagy.apm.energyalarmstarter.alarmengine.jms.model.DmpMessage;
 import com.persagy.apm.energyalarmstarter.alarmengine.service.NettyAlarmService;
 import com.persagy.apm.energyalarmstarter.alarmengine.util.StringUtil;
@@ -42,6 +47,9 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
      */
     public Map<String, Channel> socketChannelMap = new ConcurrentHashMap<>();
 
+    @Autowired
+    private AlarmServiceImpl energyAlarmService;
+
     /**
      * @description: 根据项目id获取对应的通信通道
      * @param: projectId
@@ -107,56 +115,79 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
         }
     }
 
-    /**
-     * @description: 发送全部报警定义到边缘端
-     * @param: nettyMessage
-     * @return: void
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/11/30 3:26 下午
-     * @version: V1.0
-     */
-    public void sendAllAlarmConfigs(NettyAlarmMessage nettyMessage) throws Exception {
+//    /**
+//     * @description: 发送全部报警定义到边缘端
+//     * @param: nettyMessage
+//     * @return: void
+//     * @exception:
+//     * @author: lixing
+//     * @company: Persagy Technology Co.,Ltd
+//     * @since: 2020/11/30 3:26 下午
+//     * @version: V1.0
+//     */
+//    @Deprecated
+//    public void sendAllAlarmConfigs(NettyAlarmMessage nettyMessage) throws Exception {
+//        List<JSONObject> dataList = nettyMessage.getContent();
+//        JSONObject data = dataList.get(0);
+//        String projectId = data.getString("projectId");
+//        if (StringUtils.isEmpty(projectId)) {
+//            data.put("projectId", 0);
+//        } else {
+//            data.put("projectId", projectId.split(","));
+//        }
+//
+//        data.put("userId", "system");
+//
+//        JSONArray alarmConfigs = nettyAlarmService.queryAlarmConfig(data);
+//        if (alarmConfigs == null || alarmConfigs.size() <= 0) {
+//            return;
+//        }
+//        // 查询到的报警定义按项目id分组，发送到对应的边缘端
+//        Map<String, List<Object>> groups = alarmConfigs.stream().collect(
+//                Collectors.groupingBy(alarmConfig ->
+//                        ((JSONObject) alarmConfig).getString("projectId")
+//                )
+//        );
+//
+//         /*将项目上的报警定义同步给边缘端
+//         有的边缘端可能配置了多个项目，需要将这些项目的定义先合并再发送
+//         因为边缘端如果重复接收到全量报警定义，只有最后一条生效。
+//         */
+//        Map<Channel, List<Object>> channelGroups = new HashMap<>();
+//
+//        groups.forEach((tmpProjectId, alarmConfigList) -> {
+//            Channel channel = getChannel(tmpProjectId);
+//            if (channelGroups.containsKey(channel)) {
+//                // 把要发送的报警定义追加到通道中
+//                channelGroups.get(channel).addAll(alarmConfigList);
+//            } else {
+//                channelGroups.put(channel, alarmConfigList);
+//            }
+//        });
+//
+//        channelGroups.forEach((channel, alarmConfigList) -> {
+//            sendMessage(channel, new NettyAlarmMessage(9, alarmConfigList).toString());
+//        });
+//    }
+
+    public void sendAllAlarmConditions(NettyAlarmMessage nettyMessage) throws Exception {
         List<JSONObject> dataList = nettyMessage.getContent();
         JSONObject data = dataList.get(0);
-        String projectId = data.getString("projectId");
-        if (StringUtils.isEmpty(projectId)) {
-            data.put("projectId", 0);
-        } else {
-            data.put("projectId", projectId.split(","));
+        String groupCode = data.getString("groupCode");
+        String userId = "system";
+
+        Channel channel = getChannel(null);
+        // 查询报警条件，项目报警规则，项目报警规则与设备的关联关系
+        List<AlarmCondition> conditions = energyAlarmService.queryAllAlarmCondition(userId, groupCode);
+        sendMessage(channel, new NettyAlarmMessage(NettyMsgTypeEnum.ALL_CONDITIONS, conditions).toString());
+        // 报警引擎不再配置项目信息，每个报警引擎都获取全量的报警条件
+        List<ProjectVO> projects = energyAlarmService.queryProjects(userId, groupCode);
+        // 变量项目，发送每个项目上设备与报警条件的关联关系
+        for (ProjectVO project : projects) {
+            List<ObjConditionRel> objConditionRels = energyAlarmService.queryObjAlarmConditionRel(
+                    userId, groupCode, project.getProjectId());
+            sendMessage(channel, new NettyAlarmMessage(NettyMsgTypeEnum.ALL_OBJ_CONDITION_REL, objConditionRels).toString());
         }
-
-        data.put("userId", "system");
-
-        JSONArray alarmConfigs = nettyAlarmService.queryAlarmConfig(data);
-        if (alarmConfigs == null || alarmConfigs.size() <= 0) {
-            return;
-        }
-        // 查询到的报警定义按项目id分组，发送到对应的边缘端
-        Map<String, List<Object>> groups = alarmConfigs.stream().collect(
-                Collectors.groupingBy(alarmConfig ->
-                        ((JSONObject) alarmConfig).getString("projectId")
-                )
-        );
-
-        // 全量同步报警定义，同一个边缘端只能发送一条消息，如果有多条，只有最后一条生效。
-        // 因此这里需要先将需要给边缘端发送的消息整理在一起
-        Map<Channel, List<Object>> channelGroups = new HashMap<>();
-
-        groups.forEach((tmpProjectId, alarmConfigList) -> {
-            Channel channel = getChannel(projectId);
-            if (channelGroups.containsKey(channel)) {
-                // 把要发送的报警定义追加到通道中
-                channelGroups.get(channel).addAll(alarmConfigList);
-            } else {
-                channelGroups.put(channel, alarmConfigList);
-            }
-        });
-
-        channelGroups.forEach((channel, alarmConfigList) -> {
-            sendMessage(channel, new NettyAlarmMessage(9, alarmConfigList).toString());
-        });
     }
 
     /**
@@ -182,7 +213,7 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
         List<JSONObject> records = new ArrayList<>();
         records.add(record);
         String projectId = data.getString("projectId");
-        sendMessage(projectId, new NettyAlarmMessage(8, records).toString());
+        sendMessage(projectId, new NettyAlarmMessage(NettyMsgTypeEnum.RECORD_ID, records).toString());
     }
 
     /**
@@ -202,29 +233,29 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
         nettyAlarmService.updateAlarmRecord(data);
     }
 
-    /**
-     * @description: 增量同步报警定义
-     * @param: dmpMessage
-     * @return: void
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/12/1 11:45 上午
-     * @version: V1.0
-     */
-    public void incrementSyncAlarmConfig(DmpMessage dmpMessage) throws Exception {
-        Map<String, JSONArray> changedAlarmConfigs = nettyAlarmService.queryChangedAlarmConfigs(dmpMessage);
-        JSONArray createdConfigUniques = changedAlarmConfigs.get("createdConfigUniques");
-        JSONArray deletedConfigUniques = changedAlarmConfigs.get("deletedConfigUniques");
-        if (!CollectionUtils.isEmpty(deletedConfigUniques)) {
-            // 通过netty发送给边缘端 10-云端推送删除的报警定义给边缘端(增量删除报警定义)
-            sendMessage(dmpMessage.getProjectId(), new NettyAlarmMessage(10, deletedConfigUniques).toString());
-        }
-        if (!CollectionUtils.isEmpty(createdConfigUniques)) {
-            // 通过netty发送给边缘端 7-云端推送修改的报警定义给边缘端(增量新增修改报警定义)
-            sendMessage(dmpMessage.getProjectId(), new NettyAlarmMessage(7, createdConfigUniques).toString());
-        }
-    }
+//    /**
+//     * @description: 增量同步报警定义
+//     * @param: dmpMessage
+//     * @return: void
+//     * @exception:
+//     * @author: lixing
+//     * @company: Persagy Technology Co.,Ltd
+//     * @since: 2020/12/1 11:45 上午
+//     * @version: V1.0
+//     */
+//    public void incrementSyncAlarmConfig(DmpMessage dmpMessage) throws Exception {
+//        Map<String, JSONArray> changedAlarmConfigs = nettyAlarmService.queryChangedAlarmConfigs(dmpMessage);
+//        JSONArray createdConfigUniques = changedAlarmConfigs.get("createdConfigUniques");
+//        JSONArray deletedConfigUniques = changedAlarmConfigs.get("deletedConfigUniques");
+//        if (!CollectionUtils.isEmpty(deletedConfigUniques)) {
+//            // 通过netty发送给边缘端 10-云端推送删除的报警定义给边缘端(增量删除报警定义)
+//            sendMessage(dmpMessage.getProjectId(), new NettyAlarmMessage(10, deletedConfigUniques).toString());
+//        }
+//        if (!CollectionUtils.isEmpty(createdConfigUniques)) {
+//            // 通过netty发送给边缘端 7-云端推送修改的报警定义给边缘端(增量新增修改报警定义)
+//            sendMessage(dmpMessage.getProjectId(), new NettyAlarmMessage(7, createdConfigUniques).toString());
+//        }
+//    }
 
     /**
      * @param channelHandlerContext
@@ -242,40 +273,32 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
         log.info("收到[" + channelHandlerContext.channel().remoteAddress() + "]消息:" + msg);
         try {
             NettyAlarmMessage nettyMessage = StringUtil.transferItemToDTO(msg.toString(), NettyAlarmMessage.class);
-            /* 操作类型：1-请求、2 -响应、3-通知、
-             * 4-边缘端获取报警定义、
-             * 5-边缘端主动推送报警记录、
-             * 6-边缘端主动更新报警记录状态、
-             * 7-云端推送修改的报警定义给边缘端、
-             * 9-边缘端取报警定义，云端推送给边缘端的标记
-             */
-            int opCode = nettyMessage.getOpCode();
-            List<JSONObject> dataList = nettyMessage.getContent();
-            switch (opCode) {
-                case 200:
+            switch (nettyMessage.getOpCode()) {
+                case ACCEPTED:
                     /* 接收到边缘端创建连接请求，存储连接请求的projectId和channel映射关系，
                      * 以后向边缘端发送请求时，通过projectId获取到对应的channel
                      */
                     connected(nettyMessage, channelHandlerContext);
                     break;
-                case 4:
+                case REQUEST_ALL_CONFIGS:
                     // 向边缘端全量发送报警定义
-                    sendAllAlarmConfigs(nettyMessage);
+//                    sendAllAlarmConfigs(nettyMessage);
+                    sendAllAlarmConditions(nettyMessage);
                     break;
-                case 5:
+                case CREATE_RECORD:
                     // 创建报警记录并发送报警记录id到边缘端
                     createAlarmRecordAndSendRecordId(nettyMessage);
                     break;
-                case 6:
+                case UPDATE_RECORD_STATE:
                     // 更新报警记录状态
                     updateAlarmRecord(nettyMessage);
                     break;
-                case 11:
+                case ALARM_CONTINUE:
                     // 报警仍在继续
                     alarmContinue(nettyMessage);
                     break;
                 default:
-                    log.info("边缘端发来的参数无效，参数值为：" + opCode);
+                    log.info("边缘端发来的参数无效，参数值为：" + nettyMessage.getOpCode());
                     break;
             }
 
@@ -402,4 +425,70 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
             log.error("消息通道未建立，无法发送消息！");
         }
     }
+
+    /**
+     * 同步新的报警条件
+     *
+     * @param msg 报警条件消息
+     * @author lixing
+     * @version V1.0 2021/10/26 7:51 下午
+     */
+    public void syncNewCondition(DmpMessage msg) {
+        JSONObject alarmCondition = JSONObject.parseObject(msg.getStr1());
+        sendMessage(msg.getProjectId(), new NettyAlarmMessage(
+                NettyMsgTypeEnum.NEW_CONDITION, Lists.newArrayList(alarmCondition)).toString());
+    }
+
+    /**
+     * 同步更新的报警条件
+     *
+     * @param msg 报警条件消息
+     * @author lixing
+     * @version V1.0 2021/10/26 7:51 下午
+     */
+    public void syncUpdatedCondition(DmpMessage msg) {
+        JSONObject alarmCondition = JSONObject.parseObject(msg.getStr1());
+        sendMessage(msg.getProjectId(), new NettyAlarmMessage(
+                NettyMsgTypeEnum.UPDATE_CONDITION, Lists.newArrayList(alarmCondition)).toString());
+    }
+
+    /**
+     * 同步删除的报警条件
+     *
+     * @param msg 报警条件消息
+     * @author lixing
+     * @version V1.0 2021/10/26 7:51 下午
+     */
+    public void syncDeletedCondition(DmpMessage msg) {
+        JSONObject alarmCondition = JSONObject.parseObject(msg.getStr1());
+        sendMessage(msg.getProjectId(), new NettyAlarmMessage(
+                NettyMsgTypeEnum.DELETE_CONDITION, Lists.newArrayList(alarmCondition)).toString());
+    }
+
+    /**
+     * 同步新的设备与报警条件关联关系
+     *
+     * @param msg 关联关系消息
+     * @author lixing
+     * @version V1.0 2021/10/26 7:51 下午
+     */
+    public void syncNewObjConditionRelList(DmpMessage msg) {
+        JSONArray relList = JSONObject.parseArray(msg.getStr1());
+        sendMessage(msg.getProjectId(), new NettyAlarmMessage(
+                NettyMsgTypeEnum.NEW_OBJ_CONDITION_REL, Lists.newArrayList(relList)).toString());
+    }
+
+    /**
+     * 同步删除的设备与报警条件关联关系
+     *
+     * @param msg 关联关系消息
+     * @author lixing
+     * @version V1.0 2021/10/26 7:51 下午
+     */
+    public void syncDeletedObjConditionRelList(DmpMessage msg) {
+        JSONArray relList = JSONObject.parseArray(msg.getStr1());
+        sendMessage(msg.getProjectId(), new NettyAlarmMessage(
+                NettyMsgTypeEnum.DELETE_OBJ_CONDITION_REL, Lists.newArrayList(relList)).toString());
+    }
+
 }

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/netty/NettyAlarmServer.java

@@ -16,6 +16,7 @@ import io.netty.handler.logging.LogLevel;
 import io.netty.handler.logging.LoggingHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import io.netty.util.concurrent.OrderedEventExecutor;
 import io.netty.util.concurrent.UnorderedThreadPoolEventExecutor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/netty/NettyMsgTypeEnum.java

@@ -0,0 +1,40 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.netty;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * netty的消息类型
+ *
+ * @author lixing
+ * @version V1.0 2021/10/25 3:30 下午
+ */
+@AllArgsConstructor
+public enum NettyMsgTypeEnum {
+
+    /**
+     * netty的消息类型
+     */
+    ACCEPTED(100, "已接收到消息"),
+    CONNECT(200, "建立连接，此时的source == 项目id"),
+    REQUEST_ALL_CONFIGS(10, "边缘端申请全量获取报警定义（报警条件、条件和设备的关联关系）"),
+    ALL_CONDITIONS(11, "全量报警条件"),
+    NEW_CONDITION(12, "新增报警条件"),
+    UPDATE_CONDITION(13, "更新报警条件"),
+    DELETE_CONDITION(14, "删除报警条件"),
+    ALL_OBJ_CONDITION_REL(21, "全量条件和设备的关联关系"),
+    NEW_OBJ_CONDITION_REL(22, "新增条件和设备的关联关系"),
+    DELETE_OBJ_CONDITION_REL(23, "删除条件和设备的关联关系"),
+    CREATE_RECORD(31, "创建报警记录"),
+    RECORD_ID(32, "报警记录id"),
+    UPDATE_RECORD_STATE(33, "更新报警记录状态"),
+    ALARM_CONTINUE(34, "报警仍在持续");
+
+    @Setter
+    @Getter
+    private int value;
+    @Setter
+    @Getter
+    private String desc;
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/netty/runner/NettyServerRunner.java

@@ -0,0 +1,31 @@
+package com.persagy.apm.energyalarmstarter.alarmengine.netty.runner;
+
+import com.persagy.apm.energyalarmstarter.alarmengine.netty.NettyAlarmServer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * @description: 初始化项目
+ * @author: shiliqiang
+ * @company: Persagy Technology Co.,Ltd
+ * @since: 2020/10/20 15:37
+ * @version: V1.0
+ */
+@Component
+@Slf4j
+@Order(Integer.MAX_VALUE)
+public class NettyServerRunner implements CommandLineRunner {
+	@Autowired
+	private NettyAlarmServer nettyServer;
+
+	@Override
+	public void run(String... args) throws Exception {
+		// 项目加载完成后启动netty
+		log.info("-------启动NettyAlarmServer--------");
+		nettyServer.start();
+	}
+}

AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/service/NettyAlarmService.java

@@ -40,6 +40,7 @@ public abstract class NettyAlarmService extends BaseService {
      * @since: 2020/11/30 3:25 下午
      * @version: V1.0
      */
+    @Deprecated
     public JSONArray queryAlarmConfig(JSONObject data) throws Exception {
         DmpResult<JSONArray> queryResult = alarmClient.queryAlarmConfig(getAlarmUrlParam(data), getRequestBody(data));
         JSONArray alarmConfigs = queryResult.getData();

pom.xml

@@ -24,6 +24,29 @@
         <maven.compiler.target>8</maven.compiler.target>
     </properties>
 
+    <repositories>
+        <repository>
+            <id>snapshot</id>
+            <name>snapshot Repository</name>
+            <url>http://47.93.132.139:8081/nexus/content/repositories/snapshots/</url>
+            <snapshots>
+                <enabled>true</enabled>
+                <updatePolicy>always</updatePolicy>
+            </snapshots>
+        </repository>
+        <repository>
+            <id>persagy</id>
+            <name>persagy Repository</name>
+            <url>http://47.93.132.139:8081/nexus/content/repositories/persagy/</url>
+        </repository>
+        <repository>
+            <id>integrated</id>
+            <name>releases Repository</name>
+            <url>http://47.93.132.139:8081/nexus/content/repositories/integrated/</url>
+        </repository>
+
+    </repositories>
+
     <!-- 发布maven私服 -->
     <distributionManagement>
         <snapshotRepository>