Bläddra i källkod

报警和云端通讯增加缓存处理,失败后放入缓冲队列,再次连接时候自动发送

lgy 4 år sedan
förälder
incheckning
4c6deb557b

+ 9 - 2
src/main/java/com/persagy/client/GroupNettyClient.java

@@ -1,5 +1,6 @@
 package com.persagy.client;
 
+import com.persagy.job.NettyMessageQueue;
 import com.persagy.repository.AlarmRecordRepository;
 import com.persagy.service.AlarmDefineService;
 import io.netty.bootstrap.Bootstrap;
@@ -137,8 +138,14 @@ public class GroupNettyClient {
             }
         }
     }*/
-    public void sendMessage(Object msg) {
-        channelGroup.writeAndFlush(msg);
+    public void sendMessage(String msg) throws InterruptedException {
+        log.info("给云端发送数据:[{}]",msg);
+        if(channelGroup.isWritable()){
+            channelGroup.writeAndFlush(msg);
+        }else {
+            log.warn("云端netty不可写,放入缓冲队列中[{}]",msg);
+            NettyMessageQueue.getNettyMessageQueue().produce(msg);
+        }
     }
 
 }

+ 17 - 0
src/main/java/com/persagy/client/GroupNettyClientHandler.java

@@ -8,6 +8,7 @@ import com.alibaba.fastjson.TypeReference;
 import com.persagy.entity.AlarmDefine;
 import com.persagy.entity.NettyMessage;
 import com.persagy.entity.ZktAlarmRecordDO;
+import com.persagy.job.NettyMessageQueue;
 import com.persagy.repository.AlarmRecordRepository;
 import com.persagy.service.AlarmDefineService;
 import com.persagy.utils.LockUtil;
@@ -17,6 +18,7 @@ import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.Arrays;
 import java.util.List;
@@ -81,6 +83,21 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
         nettyMessage.setContent(Arrays.asList(content));
         log.info(nettyMessage.toString());
         ctx.channel().writeAndFlush(nettyMessage.toString());
+        iniAlarmResult(ctx);
+    }
+
+    private void iniAlarmResult(ChannelHandlerContext ctx) {
+        try {
+            log.info("--initAlarmResult--");
+            String msg = NettyMessageQueue.getNettyMessageQueue().consume();
+            while (StringUtils.isNotBlank(msg)){
+                log.info("剩余报警消息令总数:{}",NettyMessageQueue.getNettyMessageQueue().size());
+                ctx.writeAndFlush(msg);
+                msg = NettyMessageQueue.getNettyMessageQueue().consume();
+            }
+        } catch (Exception e) {
+            log.error("发送报警消息失败",e);
+        }
     }
 
     /**

+ 1 - 2
src/main/java/com/persagy/client/WebSocketClientFactory.java

@@ -195,8 +195,7 @@ public class WebSocketClientFactory {
             this.setOutCallWebSocketClientHolder(webSocketClient);
             return webSocketClient;
         } catch (URISyntaxException e) {
-            e.printStackTrace();
-            log.error(e.getMessage());
+            log.error("retryOutCallWebSocketClient失败:",e);
         }
         return null;
     }

+ 3 - 1
src/main/java/com/persagy/job/ExpireAlarmQueue.java

@@ -7,6 +7,8 @@ import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * @description:报警过期,取消过期的消息队列,防止同步阻塞
+ * 报警过期,取消过期涉及到定时任务的添加和删除,多线程操作会有并发异常,单线程操作当报警很多的时候会有性能问题
+ * 增加缓冲队列
  * @author:LuoGuangyi
  * @company:PersagyTechnologyCo.,Ltd
  * @since:2020/10/27 00279:43
@@ -14,7 +16,7 @@ import java.util.concurrent.LinkedBlockingQueue;
  **/
 public class ExpireAlarmQueue {
     //队列大小
-    static final int QUEUE_MAX_SIZE   = 2000;
+    static final int QUEUE_MAX_SIZE   = 20000;
 
     static BlockingQueue<ExpireAlarmMessage> blockingQueue = new LinkedBlockingQueue<ExpireAlarmMessage>(QUEUE_MAX_SIZE);
 

+ 51 - 0
src/main/java/com/persagy/job/NettyMessageQueue.java

@@ -0,0 +1,51 @@
+package com.persagy.job;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * @description:netty通信queue
+ * 在云端与与边缘端断开连接后,报警消息,报警恢复消息,报警过期消息缓冲在边缘端的NettyMessageQueue队列中,
+ * 再次连接后,从队列发送到云端
+ * @author:LuoGuangyi
+ * @company:PersagyTechnologyCo.,Ltd
+ * @since:2020/10/27 00279:43
+ * @version:V1.0
+ **/
+public class NettyMessageQueue {
+    //队列大小
+    static final int QUEUE_MAX_SIZE   = 20000;
+
+    static BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>(QUEUE_MAX_SIZE);
+
+    /**
+     * 私有的默认构造子,保证外界无法直接实例化
+     */
+    private NettyMessageQueue(){};
+    /**
+     * 类级的内部类,也就是静态的成员式内部类,该内部类的实例与外部类的实例
+     * 没有绑定关系,而且只有被调用到才会装载,从而实现了延迟加载
+     */
+    private static class SingletonHolder{
+        /**
+         * 静态初始化器,由JVM来保证线程安全
+         */
+        private  static NettyMessageQueue queue = new NettyMessageQueue();
+    }
+    //单例队列
+    public static NettyMessageQueue getNettyMessageQueue(){
+        return SingletonHolder.queue;
+    }
+    //生产入队
+    public  void  produce(String commandResult) throws InterruptedException {
+        blockingQueue.put(commandResult);
+    }
+    //消费出队
+    public  String consume() throws InterruptedException {
+        return blockingQueue.take();
+    }
+    // 获取队列大小
+    public int size() {
+        return blockingQueue.size();
+    }
+}

+ 1 - 1
src/main/java/com/persagy/job/SpringSchedule.java

@@ -25,7 +25,7 @@ public class SpringSchedule {
     GroupNettyClient groupNettyClient;
 
     @Scheduled(cron="${alarm.get.all.alarmdefine.cron}")
-    public void allResetCron() {
+    public void allResetCron() throws InterruptedException {
         NettyMessage message = new NettyMessage(4,groupNettyClient.projectId);
         groupNettyClient.sendMessage(message.toString());
     }

+ 1 - 1
src/main/java/com/persagy/service/impl/AlarmQuartzServiceImpl.java

@@ -48,7 +48,7 @@ public class AlarmQuartzServiceImpl implements AlarmQuartzService {
                 }
             }
         } catch (Exception e) {
-            log.error("报警过期消息队列消费失败,失败原因为---->",e.getMessage());
+            log.error("报警过期消息队列消费失败",e);
         }
     }
     /**