lgy 4 роки тому
батько
коміт
dfbf9138b8

+ 1 - 1
src/main/java/com/persagy/client/GroupNettyClient.java

@@ -71,7 +71,7 @@ public class GroupNettyClient {
         GroupNettyClient groupNettyClient = this;
         try {
             // 启动辅助对象类
-            groupBootstrap.group(new NioEventLoopGroup())
+            groupBootstrap.group(new NioEventLoopGroup(5))
                     // 建立通道
                     .channel(NioSocketChannel.class)
                     // 初始化通道及进行配置

+ 2 - 3
src/main/java/com/persagy/client/GroupNettyClientHandler.java

@@ -73,9 +73,8 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
         ctx.channel().writeAndFlush(new NettyMessage<>(200,groupNettyClient.projectId).toString());
         //启动的时候发送消息,获取全部报警定义
         //{"groupCode":"wd", "projectId":"Pj123"}
-        NettyMessage nettyMessage = new NettyMessage();
+        NettyMessage nettyMessage = new NettyMessage(4,groupNettyClient.projectId);
         nettyMessage.setRemark("连接已经建立;");
-        nettyMessage.setOpCode(4);
         JSONObject content = new JSONObject();
         content.put("groupCode", groupNettyClient.groupCode);
         content.put("projectId", groupNettyClient.projectId);
@@ -115,7 +114,7 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
                         LockUtil.getInstance().lock.lock();
                         LockUtil.getInstance().setExecute(false);
                         //加个等待,保证正在执行的逻辑执行成功
-                        Thread.sleep(4000);
+                        Thread.sleep(40000);
                         alarmDefineService.listAllAlarmDefine(definesList);
                         LockUtil.getInstance().setExecute(true);
                         LockUtil.getInstance().condition.signalAll();

+ 23 - 3
src/main/java/com/persagy/client/WebSocketClientFactory.java

@@ -2,15 +2,18 @@ package com.persagy.client;
 
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.date.TimeInterval;
+import cn.hutool.core.thread.ExecutorBuilder;
 import com.persagy.service.AlarmHandleService;
 import com.persagy.utils.GZIPCompressUtil;
 import com.persagy.utils.LockUtil;
 import com.persagy.utils.StringUtil;
+import io.netty.util.concurrent.RejectedExecutionHandlers;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.java_websocket.WebSocket;
 import org.java_websocket.client.WebSocketClient;
 import org.java_websocket.handshake.ServerHandshake;
+import org.quartz.SchedulerException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Scheduled;
@@ -18,6 +21,10 @@ import org.springframework.stereotype.Component;
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -63,7 +70,12 @@ public class WebSocketClientFactory {
      */
     @Value("${project.iotid}")
     String iotProjectId;
-
+    ExecutorService executor = ExecutorBuilder.create()
+            .setCorePoolSize(5)
+            .setMaxPoolSize(10)
+            .setWorkQueue(new LinkedBlockingQueue<>(102400))
+            .setHandler(new ThreadPoolExecutor.AbortPolicy())
+            .build();
     private WebSocketClient outCallWebSocketClientHolder;
 
     public WebSocketClient getOutCallWebSocketClientHolder() {
@@ -113,7 +125,7 @@ public class WebSocketClientFactory {
                     while (!LockUtil.getInstance().isExecute()){
                         try {
                             log.warn("等待获取锁....");
-                            LockUtil.getInstance().lock.lock();
+                            LockUtil.getInstance().lock.lockInterruptibly();
                             log.warn("已经获取锁....");
 //                            try {
 //                                LockUtil.getInstance().condition.await();
@@ -127,7 +139,15 @@ public class WebSocketClientFactory {
                             LockUtil.getInstance().lock.unlock();
                         }
                     }
-                    alarmHandleService.handleIOTData(message);
+                    log.info("---[{}]",executor.toString());
+                    String finalMessage = message;
+                    executor.execute(()-> {
+                        try {
+                            alarmHandleService.handleIOTData(finalMessage);
+                        } catch (Exception e) {
+                            log.error("数据处理失败", e);
+                        }
+                    });
                 } catch (Exception e) {
                     log.error("数据处理失败", e);
                 }

+ 1 - 0
src/main/java/com/persagy/entity/NettyMessage.java

@@ -28,6 +28,7 @@ public class NettyMessage<T> {
      * 8-云端把报警记录的id推送到边缘端
      * 9-边缘端取报警定义,云端推送给边缘端的标记(全量报警定义)
      * 10-云端推送删除的报警定义给边缘端(增量删除报警定义)、
+     * 200 - 建立连接,此时的source == 项目id
      */
     @JSONField()
     private int opCode;

+ 1 - 2
src/main/java/com/persagy/job/AlarmExpireJob.java

@@ -123,9 +123,8 @@ public class AlarmExpireJob extends QuartzJobBean {
                     return;
                 }
                 log.info("报警参数为:[{}]",zktAlarmRecordDO.toString());
-                NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>();
+                NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(6,projectId);
                 nettyMessage.setStreamId(nums.getAndIncrement());
-                nettyMessage.setOpCode(6);
                 AlarmRecord message = AlarmRecord.builder()
                         .id(alarmId)
                         .state(StringUtil.getInt(state))

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

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

+ 2 - 4
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -178,8 +178,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                 //超过报警恢复设置的持续时间
                 if (DateUtils.betweenTwoTimeSecond(endTime, dateTime) >= uphold) {
                     log.error("产生一条报警恢复消息[{}]>[{}]",DateUtils.betweenTwoTimeSecond(endTime, dateTime),uphold);
-                    NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>();
-                    nettyMessage.setOpCode(6);
+                    NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(6,projectId);
                     ZktAlarmRecordDO alarmRecordDO = alarmRecordRepository.findById(AlarmInfoCache.getAlarmDefineId(alarmDefine)).orElse(new ZktAlarmRecordDO());
                     alarmRecordDO.setDefinitionId(defineId);
                     alarmRecordDO.setObjId(alarmDefine.getObjId());
@@ -320,8 +319,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                             .classCode(alarmDefine.getClassCode())
                             .createUser(systemId)
                             .build();
-                    NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>();
-                    nettyMessage.setOpCode(5);
+                    NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(5,projectId);
                     nettyMessage.setContent(Arrays.asList(alarmRecord));
                     //推送一条报警记录给远端
                     groupNettyClient.sendMessage(nettyMessage.toString());

+ 1 - 1
src/main/resources/application.yml

@@ -32,7 +32,7 @@ spring:
       initialize-schema: never  # always never
       comment-prefix: '#'
   jpa:
-    show-sql: true # 打印 SQL 。生产环境,建议关闭
+    show-sql: false # 打印 SQL 。生产环境,建议关闭
     # Hibernate 配置内容,对应 HibernateProperties 类
     hibernate:
       ddl-auto: none

+ 3 - 24
src/main/resources/logback-spring.xml

@@ -20,7 +20,7 @@ configuration 子节点为 appender、logger、root       -->
     %n:输出一个换行符号
 -->
     <!--格式化输出:%d表示日期,%thread表示线程名,%-5level:级别从左显示5个字符宽度 %msg:日志消息,%n是换行符-->
-    <property name="pattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS}  %-5level [%thread] %msg%n"/>
+    <property name="pattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS}  %-5level [%thread] %logger{36} %msg%n"/>
 
     <!--控制台-->
     <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
@@ -43,11 +43,11 @@ configuration 子节点为 appender、logger、root       -->
         </filter>
         <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
             <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
-                <maxFileSize>100MB</maxFileSize>
+                <maxFileSize>200MB</maxFileSize>
             </timeBasedFileNamingAndTriggeringPolicy>
             <!--查看   gunzip -c xx.log.gz | less-->
             <fileNamePattern>${LOG_HOME}/log.%d{yyyy-MM-dd}.log.%i.gz</fileNamePattern>
-            <maxHistory>30</maxHistory><!--保存最近30天的日志-->
+            <maxHistory>10</maxHistory><!--保存最近30天的日志-->
         </rollingPolicy>
         <encoder>
             <charset>UTF-8</charset>
@@ -88,26 +88,5 @@ configuration 子节点为 appender、logger、root       -->
     <root level="INFO">
         <appender-ref ref="stdout"/>
         <appender-ref ref="infoFile"/>
-        <!--<appender-ref ref="errorFile"/>-->
-        <!--<appender-ref ref="logstash"/>-->
     </root>
-
-    <!--为某个包单独配置logger
-
-    比如定时任务,写代码的包名为:com.seentao.task
-    步骤如下:
-    1、定义一个appender,取名为task(随意,只要下面logger引用就行了)
-    appender的配置按照需要即可
-
-
-    2、定义一个logger:
-    <logger name="com.seentao.task" level="DEBUG" additivity="false">
-      <appender-ref ref="task" />
-    </logger>
-    注意:additivity必须设置为false,这样只会交给task这个appender,否则其他appender也会打印com.seentao.task里的log信息。
-
-    3、这样,在com.seentao.task的logger就会是上面定义的logger了。
-    private static Logger logger = LoggerFactory.getLogger(Class1.class);
-    -->
-
 </configuration>