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

移除brain,适配多项目边缘端

lixing 4 роки тому
батько
коміт
7ee298696a

+ 0 - 7
pom.xml

@@ -71,13 +71,6 @@
             <artifactId>commons-lang3</artifactId>
         </dependency>
 
-        <!-- zkt-brain -->
-        <dependency>
-            <groupId>com.persagy</groupId>
-            <artifactId>zkt-brain</artifactId>
-            <version>1.0.0</version>
-        </dependency>
-
         <!--rabbitmq -->
         <dependency>
             <groupId>org.springframework.boot</groupId>

+ 1 - 1
src/main/java/com/persagy/dmp/starter/alarm/aspect/AlarmClientAspect.java

@@ -20,7 +20,7 @@ import org.springframework.stereotype.Component;
 @Aspect
 @Component
 public class AlarmClientAspect {
-    @Pointcut("execution(public * com.persagy.dmp.starter.alarm.feign.client.*.* (..))")
+    @Pointcut("execution(public * com.persagy.dmp.starter.alarm.feign.client.*.*(..))")
     public void feignPointCut() {
     }
 

+ 2 - 2
src/main/java/com/persagy/dmp/starter/alarm/communication/mq/JmsConfig.java

@@ -1,7 +1,7 @@
 package com.persagy.dmp.starter.alarm.communication.mq;
 
 import com.persagy.dmp.starter.alarm.communication.netty.NettyAlarmMsgBaseHandler;
-import com.persagy.zkt.utils.StringUtil;
+import com.persagy.dmp.starter.alarm.util.StringUtil;
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.core.*;
@@ -69,7 +69,7 @@ public class JmsConfig {
 	@RabbitListener(queues = "${dmp.alarm.queue}")
 	public void planQueues(String msg, Channel channel, Message message) throws Exception {
 		log.info("============================== Receive:" + msg);
-		DmpMessage dmpMessage = StringUtil.tranferItemToDTO(msg, DmpMessage.class);
+		DmpMessage dmpMessage = StringUtil.transferItemToDTO(msg, DmpMessage.class);
 		//报警定义变化
 		if(ALARM_CONFIGS_CHANGE.equals(dmpMessage.getType())){
 			log.info("================收到一条报警定义变化通知==============");

+ 43 - 15
src/main/java/com/persagy/dmp/starter/alarm/communication/netty/NettyAlarmMsgBaseHandler.java

@@ -1,11 +1,10 @@
 package com.persagy.dmp.starter.alarm.communication.netty;
 
-import cn.hutool.core.collection.CollectionUtil;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.persagy.dmp.starter.alarm.communication.mq.DmpMessage;
 import com.persagy.dmp.starter.alarm.service.NettyAlarmService;
-import com.persagy.zkt.utils.StringUtil;
+import com.persagy.dmp.starter.alarm.util.StringUtil;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
@@ -13,6 +12,9 @@ import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.util.concurrent.GlobalEventExecutor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.util.CollectionUtils;
 
 import java.net.SocketAddress;
 import java.util.ArrayList;
@@ -31,22 +33,39 @@ import java.util.concurrent.ConcurrentHashMap;
 @Slf4j
 public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
 
+    public static final String allProjects = "allProjects";
+
+    @Autowired
     private NettyAlarmService nettyAlarmService;
     /**
      * 装每个客户端的地址及对应的管道
      */
     public Map<String, Channel> socketChannelMap = new ConcurrentHashMap<>();
 
+    /**
+     * @description: 根据项目id获取对应的通信通道
+     * @param: projectId
+     * @return: io.netty.channel.Channel
+     * @exception:
+     * @author: lixing
+     * @company: Persagy Technology Co.,Ltd
+     * @since: 2020/12/3 3:03 下午
+     * @version: V1.0
+     */
+    private Channel getChannel(String projectId) {
+        // 项目上的消息只推送给一个边缘端来处理
+        Channel channel = socketChannelMap.get(projectId);
+        if (channel == null) {
+            channel = socketChannelMap.get(allProjects);
+        }
+        return channel;
+    }
 
     /**
      * 保留所有与服务器建立连接的channel对象
      */
     public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
 
-    public NettyAlarmMsgBaseHandler(NettyAlarmService alarmService) {
-        this.nettyAlarmService = alarmService;
-    }
-
     @Override
     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
         SocketAddress socketAddress = ctx.channel().remoteAddress();
@@ -67,7 +86,15 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
      */
     private void connected(NettyAlarmMessage nettyMessage, ChannelHandlerContext channelHandlerContext) {
         String source = nettyMessage.getSource();
-        socketChannelMap.put(source, channelHandlerContext.channel());
+        if (StringUtils.isEmpty(source)) {
+            socketChannelMap.put(allProjects, channelHandlerContext.channel());
+        } else {
+            String[] projectIds = source.split(",");
+            // 一个项目只能对应一个channel
+            for (String projectId : projectIds) {
+                socketChannelMap.put(projectId, channelHandlerContext.channel());
+            }
+        }
     }
 
     /**
@@ -80,7 +107,7 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
      * @since: 2020/11/30 3:26 下午
      * @version: V1.0
      */
-    private void sendAllAlarmConfigs(NettyAlarmMessage nettyMessage) throws Exception {
+    public void sendAllAlarmConfigs(NettyAlarmMessage nettyMessage) throws Exception {
         List<JSONObject> dataList = nettyMessage.getContent();
         JSONObject data = dataList.get(0);
         data.put("userId", "system");
@@ -103,7 +130,7 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
      * @since: 2020/11/30 3:33 下午
      * @version: V1.0
      */
-    private void createAlarmRecordAndSendRecordId(NettyAlarmMessage nettyMessage) throws Exception {
+    public void createAlarmRecordAndSendRecordId(NettyAlarmMessage nettyMessage) throws Exception {
         List<JSONObject> dataList = nettyMessage.getContent();
         JSONObject data = dataList.get(0);
         data.put("userId", "system");
@@ -129,7 +156,7 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
      * @since: 2020/11/30 4:13 下午
      * @version: V1.0
      */
-    private void updateAlarmRecord(NettyAlarmMessage message) throws Exception {
+    public void updateAlarmRecord(NettyAlarmMessage message) throws Exception {
         List<JSONObject> dataList = message.getContent();
         JSONObject data = dataList.get(0);
         data.put("userId", "system");
@@ -150,11 +177,11 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
         Map<String, JSONArray> changedAlarmConfigs = nettyAlarmService.queryChangedAlarmConfigs(dmpMessage);
         JSONArray createdConfigUniques = changedAlarmConfigs.get("createdConfigUniques");
         JSONArray deletedConfigUniques = changedAlarmConfigs.get("deletedConfigUniques");
-        if (CollectionUtil.isNotEmpty(deletedConfigUniques)) {
+        if (!CollectionUtils.isEmpty(deletedConfigUniques)) {
             // 通过netty发送给边缘端 10-云端推送删除的报警定义给边缘端(增量删除报警定义)
             sendMessage(dmpMessage.getProjectId(), new NettyAlarmMessage(10, deletedConfigUniques).toString());
         }
-        if (CollectionUtil.isNotEmpty(createdConfigUniques)) {
+        if (!CollectionUtils.isEmpty(createdConfigUniques)) {
             // 通过netty发送给边缘端 7-云端推送修改的报警定义给边缘端(增量新增修改报警定义)
             sendMessage(dmpMessage.getProjectId(), new NettyAlarmMessage(7, createdConfigUniques).toString());
         }
@@ -175,7 +202,7 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
     public void channelRead(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
         log.info("收到[" + channelHandlerContext.channel().remoteAddress() + "]消息:" + msg);
         try {
-            NettyAlarmMessage nettyMessage = StringUtil.tranferItemToDTO(msg.toString(), NettyAlarmMessage.class);
+            NettyAlarmMessage nettyMessage = StringUtil.transferItemToDTO(msg.toString(), NettyAlarmMessage.class);
             /* 操作类型:1-请求、2 -响应、3-通知、
              * 4-边缘端获取报警定义、
              * 5-边缘端主动推送报警记录、
@@ -294,8 +321,9 @@ public class NettyAlarmMsgBaseHandler extends ChannelInboundHandlerAdapter {
      * @Description: 服务端给某个客户端发送消息
      */
     public void sendMessage(String projectId, String msg) {
-        if (socketChannelMap.containsKey(projectId)) {
-            socketChannelMap.get(projectId).writeAndFlush(msg);
+        Channel channel = getChannel(projectId);
+        if (channel != null) {
+            channel.writeAndFlush(msg);
         } else {
             log.info("...projectId[{}]未建立连接,无法发送!", projectId);
         }

+ 101 - 4
src/main/java/com/persagy/dmp/starter/alarm/communication/websocket/AlarmWebSocketCache.java

@@ -1,10 +1,13 @@
 package com.persagy.dmp.starter.alarm.communication.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
@@ -15,17 +18,61 @@ import java.util.concurrent.ConcurrentHashMap;
  */
 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) {
+            // 结果集中一定包含接收所有项目消息的通道
+            resultSet.addAll(projectSessionIds.get(AlarmWebSocketCache.allProjects));
+            if (StringUtils.isNotEmpty(projectId)) {
+                resultSet.addAll(projectSessionIds.get(projectId));
+            }
+        }
+        return resultSet;
+    }
+
+
+    /**
+     * 所有在线的客户端
+     * projectId:<sessionId1,sessionId2>
+     */
+    private static Map<String, Set<String>> projectSessionIds = new ConcurrentHashMap<>();
+
+
+    /**
      * 获取连接中的所有客户端
      *
      * @return
      */
-    public static Set<Session> getClients() {
-    	return new HashSet<>(clients.values());
+    public static Session getClient(String sessionId) {
+        return clients.get(sessionId);
+    }
+
+    /**
+     * 获取连接中的所有客户端
+     *
+     * @return
+     */
+    public static Map<String, Session> getClients() {
+        return clients;
     }
 
     /**
@@ -33,15 +80,65 @@ public class AlarmWebSocketCache {
      *
      * @param session
      */
-    public static void addClient(Session 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 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);
     }
 

+ 41 - 12
src/main/java/com/persagy/dmp/starter/alarm/communication/websocket/AlarmWebSocketServer.java

@@ -1,11 +1,13 @@
 package com.persagy.dmp.starter.alarm.communication.websocket;
 
-import cn.hutool.core.collection.CollectionUtil;
 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;
 
 /**
@@ -16,7 +18,7 @@ import java.util.Set;
  * @since: 2020/11/30 6:47 下午
  * @version: V1.0
  */
-@ServerEndpoint(value = "/webSocketServer")
+@ServerEndpoint(value = "/websocket/iot")
 @Component
 @Slf4j
 public class AlarmWebSocketServer {
@@ -25,7 +27,30 @@ public class AlarmWebSocketServer {
     public void onOpen(Session session) {
         log.info("有新的客户端建立连接,编号: " + session.getId());
         //将新用户存入在线的组
-        AlarmWebSocketCache.addClient(session);
+        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);
     }
 
     /**
@@ -38,7 +63,8 @@ public class AlarmWebSocketServer {
         String clientId = session.getId();
         log.info("客户端断开连接,编号:" + clientId);
         //将掉线的用户移除在线的组里
-        AlarmWebSocketCache.removeClient(clientId);
+        String projectIdStr = getProjectIdStr(session);
+        AlarmWebSocketCache.removeClient(projectIdStr, clientId);
     }
 
     /**
@@ -51,7 +77,8 @@ public class AlarmWebSocketServer {
         String id = "";
         if (null != session) {
             id = session.getId();
-            AlarmWebSocketCache.removeClient(id);
+            String projectIdStr = getProjectIdStr(session);
+            AlarmWebSocketCache.removeClient(projectIdStr, id);
         }
         log.info("客户端{}出错 ", id);
     }
@@ -73,7 +100,8 @@ public class AlarmWebSocketServer {
 
 
     /**
-     * @param msg 消息
+     * @param projectId 项目ID
+     * @param msg       消息
      * @description: 发送消息
      * @return: void
      * @exception:
@@ -82,12 +110,13 @@ public class AlarmWebSocketServer {
      * @since: 2020/10/21 22:30
      * @version: V1.0
      */
-    public static void sendMsgToClients(String msg) throws Exception {
-        Set<Session> clients = AlarmWebSocketCache.getClients();
-        synchronized (clients) {
-            if (CollectionUtil.isNotEmpty(clients)) {
-                for (Session session : clients) {
-                    if (null != session && session.isOpen()) {
+    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);
                     }

+ 9 - 10
src/main/java/com/persagy/dmp/starter/alarm/service/NettyAlarmService.java

@@ -1,6 +1,5 @@
 package com.persagy.dmp.starter.alarm.service;
 
-import cn.hutool.core.collection.CollectionUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
@@ -39,7 +38,7 @@ public abstract class NettyAlarmService {
      * @since: 2020/11/30 3:14 下午
      * @version: V1.0
      */
-    private void checkRequestParam(JSONObject obj) throws Exception {
+    public void checkRequestParam(JSONObject obj) throws Exception {
         String projectId = obj.getString("projectId");
         String userId = obj.getString("userId");
         String groupCode = obj.getString("groupCode");
@@ -62,7 +61,7 @@ public abstract class NettyAlarmService {
      * @since: 2020/11/30 3:14 下午
      * @version: V1.0
      */
-    private AlarmUrlParam getAlarmUrlParam(JSONObject obj) throws Exception {
+    public AlarmUrlParam getAlarmUrlParam(JSONObject obj) throws Exception {
         this.checkRequestParam(obj);
 
         return new AlarmUrlParam(
@@ -83,7 +82,7 @@ public abstract class NettyAlarmService {
      * @since: 2020/11/30 3:14 下午
      * @version: V1.0
      */
-    private JSONObject getRequestBody(JSONObject obj) throws Exception {
+    public JSONObject getRequestBody(JSONObject obj) throws Exception {
         JSONObject requestBody = new JSONObject();
         JSONObject criteria = (JSONObject) obj.clone();
         criteria.remove("appId");
@@ -144,7 +143,7 @@ public abstract class NettyAlarmService {
      * @since: 2020/12/1 2:32 下午
      * @version: V1.0
      */
-    private void initAlarmConfigInfoCodes(JSONArray alarmConfigs) throws Exception {
+    public void initAlarmConfigInfoCodes(JSONArray alarmConfigs) throws Exception {
         if (alarmConfigs == null) {
             return;
         }
@@ -274,9 +273,9 @@ public abstract class NettyAlarmService {
         JSONArray createdConfigUniques = exts.getJSONArray("createdConfigUniques");
         //修改报警定义
         JSONArray updatedConfigUniques = exts.getJSONArray("updatedConfigUniques");
-        createdConfigUniques = CollectionUtil.isEmpty(createdConfigUniques) ? new JSONArray() : createdConfigUniques;
-        deletedConfigUniques = CollectionUtil.isEmpty(deletedConfigUniques) ? new JSONArray() : deletedConfigUniques;
-        updatedConfigUniques = CollectionUtil.isEmpty(updatedConfigUniques) ? new JSONArray() : updatedConfigUniques;
+        createdConfigUniques = CollectionUtils.isEmpty(createdConfigUniques) ? new JSONArray() : createdConfigUniques;
+        deletedConfigUniques = CollectionUtils.isEmpty(deletedConfigUniques) ? new JSONArray() : deletedConfigUniques;
+        updatedConfigUniques = CollectionUtils.isEmpty(updatedConfigUniques) ? new JSONArray() : updatedConfigUniques;
         createdConfigUniques.addAll(updatedConfigUniques);
         List<String> defineList = createdConfigUniques.stream().map(
                 p -> JSONObject.parseObject(JSONObject.toJSONString(p))
@@ -298,7 +297,7 @@ public abstract class NettyAlarmService {
             data.put("userId", "system");
             DmpResult<JSONArray> alarmConfigQueryResult = alarmService.queryAlarmConfig(getAlarmUrlParam(data), getRequestBody(data));
             JSONArray tmpAlarmConfigArr = alarmConfigQueryResult.getData();
-            if (CollectionUtil.isNotEmpty(tmpAlarmConfigArr)) {
+            if (!CollectionUtils.isEmpty(tmpAlarmConfigArr)) {
                 // 报警定义中的信息点转换为表号、功能号
                 initAlarmConfigInfoCodes(tmpAlarmConfigArr);
                 alarmConfigArr.addAll(tmpAlarmConfigArr);
@@ -321,7 +320,7 @@ public abstract class NettyAlarmService {
      * @since: 2020/12/1 11:54 上午
      * @version: V1.0
      */
-    private String getAlarmConfigDefineId(JSONObject obj) {
+    public String getAlarmConfigDefineId(JSONObject obj) {
         return obj.getString("itemCode") + "" + obj.getString("objId");
     }
 }

+ 20 - 0
src/main/java/com/persagy/dmp/starter/alarm/util/StringUtil.java

@@ -0,0 +1,20 @@
+package com.persagy.dmp.starter.alarm.util;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * @description:
+ * @author: lixing
+ * @company: Persagy Technology Co.,Ltd
+ * @since: 2020/12/3 4:40 下午
+ * @version: V1.0
+ **/
+public class StringUtil {
+    public static <T, R> T transferItemToDTO(String content, Class<T> t) throws Exception {
+        if (StringUtils.isNotBlank(content)) {
+            return JSONObject.parseObject(content, t);
+        }
+        return t.newInstance();
+    }
+}