Browse Source

Merge branch 'v2.0.0' of http://39.106.8.246:3003/apm/energy-alarm-starter into v2.0.0

lixing 3 years ago
parent
commit
9968bbf6bd

+ 1 - 1
AlarmDataStarter/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>energy-alarm-starter</artifactId>
         <groupId>com.persagy.apm</groupId>
-        <version>v1.0.0-SNAPSHOT</version>
+        <version>v2.0.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

+ 1 - 1
AlarmEngineStarter/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>energy-alarm-starter</artifactId>
         <groupId>com.persagy.apm</groupId>
-        <version>v1.0.0-SNAPSHOT</version>
+        <version>v2.0.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

+ 1 - 4
AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/AlarmEngineMsgHandler.java

@@ -52,10 +52,7 @@ public class AlarmEngineMsgHandler extends ChannelInboundHandlerAdapter {
     public void createAlarmRecordAndSendRecordId(DmpMessage dmpMessage) throws Exception {
         JSONObject data = dmpMessage.getExts();
         data.put("userId", "system");
-        String alarmRecordId = alarmRecordMsgHandler.createAlarm(data);
-        String objId = data.getString("objId");
-        String itemCode = data.getString("itemCode");
-        // TODO: 2021/11/16  把报警id放入redis缓存
+        alarmRecordMsgHandler.createAlarm(data);
     }
 
     /**

+ 51 - 0
AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/JmsAlarmRecordEnum.java

@@ -0,0 +1,51 @@
+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;
+
+/**
+ * 报警记录
+ */
+@SwaggerDisplayEnum
+@AllArgsConstructor
+public enum JmsAlarmRecordEnum {
+    /**
+     * 字典类型
+     */
+    NEW_ALARM_RECORD("new_alarmRecord", "新增报警记录"),
+    UPDATE_ALARM_RECORD("update_alarmRecord", "更新报警记录"),
+    CONDITION_ALARM("continue_alarm", "报警持续处理");
+
+    @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 JmsAlarmRecordEnum getByType(String type) {
+        if (StringUtils.isBlank(type)) {
+            return null;
+        }
+        JmsAlarmRecordEnum[] enums = JmsAlarmRecordEnum.values();
+        for (JmsAlarmRecordEnum item : enums) {
+            if (type.equals(item.getType())) {
+                return item;
+            }
+        }
+        return null;
+    }
+}
+
+

+ 61 - 1
AlarmEngineStarter/src/main/java/com/persagy/apm/energyalarmstarter/alarmengine/jms/JmsConfig.java

@@ -1,14 +1,17 @@
 package com.persagy.apm.energyalarmstarter.alarmengine.jms;
 
+import com.alibaba.fastjson.JSONObject;
 import com.persagy.apm.energyalarmstarter.alarmengine.jms.model.DmpMessage;
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
 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.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.RedisTemplate;
 
 /**
  * @description:报警定义消息通知
@@ -27,6 +30,9 @@ public class JmsConfig {
     @Autowired
     private AlarmEngineMsgHandler msgHandler;
 
+    @Autowired
+    private RedisTemplate redisTemplate;
+
     /**
      * 交换机
      */
@@ -43,6 +49,11 @@ public class JmsConfig {
     private final String objConditionRelRoutingKey = "obj-condition-rel-routing-key";
 
     /**
+     * 报警记录 路由键
+     */
+    private final String alarmRecordRoutingKey = "alarm-msg-routing-key";
+
+    /**
      * 报警条件队列
      */
     private final String conditionQueue = "condition-queue";
@@ -51,6 +62,10 @@ public class JmsConfig {
      * 设备与报警条件关联关系队列
      */
     private final String objConditionRelQueue = "obj-condition-rel-queue";
+    /**
+     * 报警记录队列
+     */
+    private final String alarmRecordQueue = "alarm-msg-queue";
 
     @Bean
     public Queue conditionQueue() {
@@ -63,6 +78,11 @@ public class JmsConfig {
     }
 
     @Bean
+    public Queue alarmRecordQueue() {
+        return new Queue(alarmRecordQueue, true);
+    }
+
+    @Bean
     public TopicExchange alarmServiceExchange() {
         return new TopicExchange(exchange);
     }
@@ -77,6 +97,11 @@ public class JmsConfig {
         return BindingBuilder.bind(objConditionRelQueue()).to(alarmServiceExchange()).with(objConditionRelRoutingKey);
     }
 
+    @Bean
+    public Binding alarmRecordBinding() {
+        return BindingBuilder.bind(alarmRecordQueue()).to(alarmServiceExchange()).with(alarmRecordRoutingKey);
+    }
+
     @RabbitHandler
     @RabbitListener(queues = {conditionQueue, objConditionRelQueue})
     public void dealMsg(DmpMessage msg, Channel channel, Message message) throws Exception {
@@ -98,8 +123,43 @@ public class JmsConfig {
                 msgHandler.syncDeletedObjConditionRelList(msg);
             }
         } catch (Exception e) {
-            log.error("消息消费失败,{}", e.getMessage());
+            log.error("报警条件消息消费失败,{}", e.getMessage());
         }
         channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
     }
+
+    @RabbitHandler
+    @RabbitListener(queues = {alarmRecordQueue})
+    public void setAlarmRecordMsg(DmpMessage msg, Channel channel, Message message) throws Exception {
+        try {
+            log.debug("============================== Receive:" + msg);
+            if (JmsAlarmRecordEnum.NEW_ALARM_RECORD.getType().equals(msg.getType())) {
+                msgHandler.createAlarmRecordAndSendRecordId(msg);
+            }
+            if (JmsAlarmRecordEnum.UPDATE_ALARM_RECORD.getType().equals(msg.getType())) {
+                msgHandler.updateAlarmRecord(msg);
+            }
+            if (JmsAlarmRecordEnum.CONDITION_ALARM.getType().equals(msg.getType())) {
+                msgHandler.alarmContinue(msg);
+            }
+            //解锁
+            JSONObject data = msg.getExts();
+            unlock(data.getString("id"));
+        } catch (Exception e) {
+            log.error("报警记录消息消费失败,{}", e.getMessage());
+        }
+        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+    }
+
+
+    public void unlock(String key) {
+        try {
+            String currentValue = String.valueOf(redisTemplate.opsForValue().get(key));
+            if (!StringUtils.isEmpty(currentValue)) {
+                redisTemplate.opsForValue().getOperations().delete(key);
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
 }

+ 0 - 56
CollectDataStarter/pom.xml

@@ -1,56 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>energy-alarm-starter</artifactId>
-        <groupId>com.persagy.apm</groupId>
-        <version>v1.0.0-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>collect-data-starter</artifactId>
-
-    <properties>
-        <maven.compiler.source>8</maven.compiler.source>
-        <maven.compiler.target>8</maven.compiler.target>
-    </properties>
-
-    <dependencies>
-        <!-- websocket -->
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-websocket</artifactId>
-<!--            <exclusions>-->
-<!--                <exclusion>-->
-<!--                    <artifactId>spring-boot-starter-logging</artifactId>-->
-<!--                    <groupId>org.springframework.boot</groupId>-->
-<!--                </exclusion>-->
-<!--            </exclusions>-->
-        </dependency>
-
-        <!-- starter 标配 -->
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-configuration-processor</artifactId>
-            <optional>true</optional>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-autoconfigure</artifactId>
-        </dependency>
-
-        <!-- lombok -->
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-            <optional>true</optional>
-        </dependency>
-
-        <!-- lang -->
-        <dependency>
-            <groupId>org.apache.commons</groupId>
-            <artifactId>commons-lang3</artifactId>
-        </dependency>
-    </dependencies>
-</project>

+ 0 - 17
CollectDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/collectdata/AutoConfiguration.java

@@ -1,17 +0,0 @@
-package com.persagy.apm.energyalarmstarter.collectdata;
-
-import org.springframework.context.annotation.ComponentScan;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @description: 配置类
- * @author: lixing
- * @company: Persagy Technology Co.,Ltd
- * @since: 2020/11/27 4:51 下午
- * @version: V1.0
- */
-@Configuration
-@ComponentScan(value = "com.persagy.apm.energyalarmstarter.collectdata")
-public class AutoConfiguration {
-
-}

+ 0 - 149
CollectDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/collectdata/websocket/AlarmWebSocketCache.java

@@ -1,149 +0,0 @@
-package com.persagy.apm.energyalarmstarter.collectdata.websocket;
-
-import org.apache.commons.lang3.StringUtils;
-
-import javax.websocket.Session;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArraySet;
-
-/**
- * @description: webSocket缓存,存储通信session
- * @author: lixing
- * @company: Persagy Technology Co.,Ltd
- * @since: 2020/11/30 7:07 下午
- * @version: V1.0
- */
-public class AlarmWebSocketCache {
-    /**
-     * 所有项目的标志
-     */
-    public static final String allProjects = "allProjects";
-    /**
-     * 所有在线的客户端
-     */
-    private static Map<String, Session> clients = new ConcurrentHashMap<>();
-
-    /**
-     * @description: 根据项目获取通道
-     * @param: projectId
-     * @return: java.util.Set<java.lang.String>
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/12/3 2:56 下午
-     * @version: V1.0
-     */
-    public static Set<String> getProjectSessionIds(String projectId) {
-        // 如果多个边缘端配置的projectId有重复,这里不知道报警定义到底发送到了哪个边缘端,所以这里发消息要发给所有边缘端。
-        Set<String> resultSet = new HashSet<>();
-        synchronized (resultSet) {
-            // 结果集中一定包含接收所有项目消息的通道
-            if (projectSessionIds.get(AlarmWebSocketCache.allProjects) != null) {
-                resultSet.addAll(projectSessionIds.get(AlarmWebSocketCache.allProjects));
-            }
-            if (StringUtils.isNotEmpty(projectId)) {
-                if (projectSessionIds.get(projectId) != null) {
-                    resultSet.addAll(projectSessionIds.get(projectId));
-                }
-            }
-        }
-        return resultSet;
-    }
-
-
-    /**
-     * 所有在线的客户端
-     * projectId:<sessionId1,sessionId2>
-     */
-    private static Map<String, Set<String>> projectSessionIds = new ConcurrentHashMap<>();
-
-
-    /**
-     * 获取连接中的所有客户端
-     *
-     * @return
-     */
-    public static Session getClient(String sessionId) {
-        return clients.get(sessionId);
-    }
-
-    /**
-     * 获取连接中的所有客户端
-     *
-     * @return
-     */
-    public static Map<String, Session> getClients() {
-        return clients;
-    }
-
-    /**
-     * 增加客户端
-     *
-     * @param session
-     */
-    public static void addClient(String projectIdStr, Session session) {
-        if (StringUtils.isEmpty(projectIdStr)) {
-            addProjectSessionId(allProjects, session.getId());
-        } else {
-            String[] projectIds = projectIdStr.split(",");
-            for (String projectId : projectIds) {
-                addProjectSessionId(projectId, session.getId());
-            }
-        }
-        clients.put(session.getId(), session);
-    }
-
-    /**
-     * @description: 将sessionId添加到项目下
-     * @param: projectId
-     * @param: session
-     * @return: void
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/12/2 6:25 下午
-     * @version: V1.0
-     */
-    private static void addProjectSessionId(String projectId, String sessionId) {
-        Set<String> sessionIds = projectSessionIds.getOrDefault(projectId, new CopyOnWriteArraySet());
-        sessionIds.add(sessionId);
-        projectSessionIds.put(projectId, sessionIds);
-    }
-
-    /**
-     * @description: 将sessionId从项目下移除
-     * @param: projectId
-     * @param: session
-     * @return: void
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/12/2 6:25 下午
-     * @version: V1.0
-     */
-    private static void removeProjectSessionId(String projectId, String sessionId) {
-        Set<String> sessionIds = projectSessionIds.getOrDefault(projectId, new CopyOnWriteArraySet());
-        sessionIds.remove(sessionId);
-    }
-
-    /**
-     * 删除客户端
-     *
-     * @param sessionId 客户端标识
-     */
-    public static void removeClient(String projectIdStr, String sessionId) {
-        if (StringUtils.isEmpty(projectIdStr)) {
-            removeProjectSessionId(allProjects, sessionId);
-        } else {
-            String[] projectIds = projectIdStr.split(",");
-            for (String projectId : projectIds) {
-                removeProjectSessionId(projectId, sessionId);
-            }
-        }
-        clients.remove(sessionId);
-    }
-
-}

+ 0 - 127
CollectDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/collectdata/websocket/AlarmWebSocketServer.java

@@ -1,127 +0,0 @@
-package com.persagy.apm.energyalarmstarter.collectdata.websocket;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-import org.springframework.util.CollectionUtils;
-
-import javax.websocket.*;
-import javax.websocket.server.ServerEndpoint;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * @description: websocket服务端
- * 由于是websocket 所以原本是@RestController的http形式直接替换成@ServerEndpoint即可,作用是一样的 就是指定一个地址表示定义一个websocket的Server端
- * @author: lixing
- * @company: Persagy Technology Co.,Ltd
- * @since: 2020/11/30 6:47 下午
- * @version: V1.0
- */
-@ServerEndpoint(value = "/websocket/iot")
-@Component
-@Slf4j
-public class AlarmWebSocketServer {
-
-    @OnOpen
-    public void onOpen(Session session) {
-        log.info("有新的客户端建立连接,编号: " + session.getId());
-        //将新用户存入在线的组
-        String projectIdStr = getProjectIdStr(session);
-        AlarmWebSocketCache.addClient(projectIdStr, session);
-    }
-
-    /**
-     * @description: 获取查询条件中的projectId
-     * @param: session
-     * @return: java.lang.String
-     * @exception:
-     * @author: lixing
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/12/2 6:53 下午
-     * @version: V1.0
-     */
-    private String getProjectIdStr(Session session) {
-        Map<String, List<String>> requestParameterMap = session.getRequestParameterMap();
-        if (requestParameterMap == null) {
-            return null;
-        }
-        List<String> projectIds = requestParameterMap.get("projectId");
-        if (CollectionUtils.isEmpty(projectIds)) {
-            return null;
-        }
-        return projectIds.get(0);
-    }
-
-    /**
-     * 客户端关闭
-     *
-     * @param session session
-     */
-    @OnClose
-    public void onClose(Session session) {
-        String clientId = session.getId();
-        log.info("客户端断开连接,编号:" + clientId);
-        //将掉线的用户移除在线的组里
-        String projectIdStr = getProjectIdStr(session);
-        AlarmWebSocketCache.removeClient(projectIdStr, clientId);
-    }
-
-    /**
-     * 发生错误
-     *
-     * @param throwable e
-     */
-    @OnError
-    public void onError(Session session, Throwable throwable) {
-        String id = "";
-        if (null != session) {
-            id = session.getId();
-            String projectIdStr = getProjectIdStr(session);
-            AlarmWebSocketCache.removeClient(projectIdStr, id);
-        }
-        log.info("客户端{}出错 ", id);
-    }
-
-    /**
-     * 收到客户端发来消息
-     *
-     * @param message 消息对象,格式:point,org,1101010001,10001-101,10001-102,10001-103,10001-104
-     */
-    @OnMessage
-    public void onMessage(String message, Session session) {
-        String clientId = session.getId();
-        log.info("收到客户端{}的数据请求消息:{}", clientId, message);
-        if ("closeSession".equals(message)) {
-            onClose(session);
-            return;
-        }
-    }
-
-
-    /**
-     * @param projectId 项目ID
-     * @param msg       消息
-     * @description: 发送消息
-     * @return: void
-     * @exception:
-     * @author: shiliqiang
-     * @company: Persagy Technology Co.,Ltd
-     * @since: 2020/10/21 22:30
-     * @version: V1.0
-     */
-    public static void sendMsgToClients(String projectId, String msg) throws Exception {
-        Set<String> projectSessionIds = AlarmWebSocketCache.getProjectSessionIds(projectId);
-        if (!CollectionUtils.isEmpty(projectSessionIds)) {
-            for (String sessionId : projectSessionIds) {
-                Session session = AlarmWebSocketCache.getClient(sessionId);
-                if (null != session && session.isOpen()) {
-                    synchronized (session) {
-                        //同步发送
-                        session.getBasicRemote().sendText(msg);
-                    }
-                }
-            }
-        }
-    }
-}

+ 0 - 20
CollectDataStarter/src/main/java/com/persagy/apm/energyalarmstarter/collectdata/websocket/AlarmWebsocketConfiguration.java

@@ -1,20 +0,0 @@
-package com.persagy.apm.energyalarmstarter.collectdata.websocket;
-
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.web.socket.server.standard.ServerEndpointExporter;
-
-/**
- * @description: websocket配置类
- * @author: lixing
- * @company: Persagy Technology Co.,Ltd
- * @since: 2020/11/30 6:49 下午
- * @version: V1.0
- */
-@Configuration
-public class AlarmWebsocketConfiguration {
-	@Bean
-    public ServerEndpointExporter serverEndpointExporter() {
-        return new ServerEndpointExporter();
-    }	
-}

+ 1 - 2
pom.xml

@@ -12,11 +12,10 @@
     <groupId>com.persagy.apm</groupId>
     <artifactId>energy-alarm-starter</artifactId>
     <packaging>pom</packaging>
-    <version>v1.0.0-SNAPSHOT</version>
+    <version>v2.0.0-SNAPSHOT</version>
     <modules>
         <module>AlarmDataStarter</module>
         <module>AlarmEngineStarter</module>
-        <module>CollectDataStarter</module>
     </modules>
 
     <properties>