Selaa lähdekoodia

完善一下注释和小的逻辑

lgy 4 vuotta sitten
vanhempi
commit
629c0b6f9d

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

@@ -95,7 +95,6 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
          log.info("Client received: {}",msg);
         try {
             TimeInterval timer = DateUtil.timer();
-            LockUtil.getInstance().setExecute(false);
             handlerMsg(ctx, msg);
             log.info("处理消息时间[{}]",timer.interval());
         } catch (Exception e) {
@@ -141,6 +140,8 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
                     String defineId = parseObject.getString("itemCode") + "-" + parseObject.getString("objId");
                     ZktAlarmRecordDO zktAlarmRecordDO = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO());
                     zktAlarmRecordDO.setDefinitionId(defineId);
+                    zktAlarmRecordDO.setItemCode(parseObject.getString("itemCode"));
+                    zktAlarmRecordDO.setObjId(parseObject.getString("objId"));
                     zktAlarmRecordDO.setAlarmId(parseObject.getString("id"));
                     alarmRecordRepository.save(zktAlarmRecordDO);
                 }

+ 4 - 2
src/main/java/com/persagy/entity/NettyMessage.java

@@ -21,8 +21,10 @@ public class NettyMessage<T> {
     private int version = 1;
     /**
      * 操作类型:1-请求、2 -响应、3-通知、
-     * 4-边缘端获取报警定义、5-边缘端主动推送报警记录、
-     * 6-边缘端主动更新报警记录状态、7-云端推送修改的报警定义给边缘端(增量报警定义)、
+     * 4-边缘端获取报警定义、
+     * 5-边缘端主动推送报警记录、
+     * 6-边缘端主动更新报警记录状态、
+     * 7-云端推送修改的报警定义给边缘端(增量报警定义)、
      * 8-云端把报警记录的id推送到边缘端
      * 9-边缘端取报警定义,云端推送给边缘端的标记(全量报警定义)
      */

+ 9 - 2
src/main/java/com/persagy/init/InitRunner.java

@@ -4,6 +4,7 @@ package com.persagy.init;
 import com.googlecode.aviator.AviatorEvaluator;
 import com.persagy.client.GroupNettyClient;
 import com.persagy.client.WebSocketClientFactory;
+import com.persagy.service.AlarmQuartzService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
@@ -20,14 +21,20 @@ public class InitRunner implements CommandLineRunner {
 	private WebSocketClientFactory webSocketClientFactory;
 	@Autowired
 	private GroupNettyClient groupNettyClient;
-
+	@Autowired
+	AlarmQuartzService alarmQuartzService;
 
 
 	@Override
 	public void run(String... args) throws Exception {
 		//5.0 开始引入了 LRU 缓存,可指定缓存的表达式个数,比如设置为最大 1 万个缓存结果:
 		AviatorEvaluator.getInstance().useLRUExpressionCache(10000);
-		webSocketClientFactory.retryOutCallWebSocketClient();
+		//启动netty客户端,接受云端数据
 		groupNettyClient.start();
+		//启动websocket,接受IOT采集庶数据
+		webSocketClientFactory.retryOutCallWebSocketClient();
+		//
+		alarmQuartzService.initAlarmMessagehandler();
+
 	}
 }

+ 19 - 0
src/main/java/com/persagy/job/AlarmExpireJob.java

@@ -4,7 +4,9 @@ import cn.hutool.core.date.DatePattern;
 import cn.hutool.core.date.DateTime;
 import cn.hutool.core.date.DateUtil;
 import com.alibaba.fastjson.JSONObject;
+import com.persagy.cache.AlarmInfoCache;
 import com.persagy.client.GroupNettyClient;
+import com.persagy.entity.AlarmState;
 import com.persagy.entity.NettyMessage;
 import com.persagy.entity.ZktAlarmRecordDO;
 import com.persagy.repository.AlarmRecordRepository;
@@ -31,6 +33,8 @@ public class AlarmExpireJob extends QuartzJobBean {
     private GroupNettyClient groupNettyClient;
     @Autowired
     AlarmRecordRepository alarmRecordRepository;
+    @Autowired
+    AlarmInfoCache alarmInfoCache;
     /**
      * 报警记录信息详情
      */
@@ -44,6 +48,10 @@ public class AlarmExpireJob extends QuartzJobBean {
      */
     private String expireTime;
     /**
+     * 过期时间
+     */
+    private String defineId;
+    /**
      * 集团编码
      */
     @Value("${group.code}")
@@ -84,6 +92,9 @@ public class AlarmExpireJob extends QuartzJobBean {
                 nettyMessage.setContent(Arrays.asList(result));
                 //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123"}
                 groupNettyClient.sendMessage(nettyMessage.toString());
+                //已经过期的时候删除掉这条报警定义了,保证不会再次产生报警
+                AlarmState alarmState = new AlarmState(defineId);
+                alarmInfoCache.setAlarmState(defineId, alarmState);
                 if(alarmRecordRepository.existsById(zktAlarmRecordDO.getDefinitionId())){
                     alarmRecordRepository.deleteById(zktAlarmRecordDO.getDefinitionId());
                 }
@@ -141,4 +152,12 @@ public class AlarmExpireJob extends QuartzJobBean {
     public void setExpireTime(String expireTime) {
         this.expireTime = expireTime;
     }
+
+    public String getDefineId() {
+        return defineId;
+    }
+
+    public void setDefineId(String defineId) {
+        this.defineId = defineId;
+    }
 }

+ 2 - 0
src/main/java/com/persagy/service/AlarmQuartzService.java

@@ -13,6 +13,8 @@ import java.util.Date;
  * @version:V1.0
  **/
 public interface AlarmQuartzService {
+    void initAlarmMessagehandler();
+
     String addExpireJob(Date startTime, String jobName, String jobGroupName, JobDataMap jobDataMap) throws SchedulerException;
 
     String deleteExpireJob(String jobName, String jobGroupName) throws SchedulerException;

+ 47 - 30
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -62,7 +62,16 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
      */
     @Value("${project.id}")
     String projectId;
-
+    /**
+     * @description:处理iot采集数据
+     * @exception:
+     * @author: LuoGuangyi
+     * @company: Persagy Technology Co.,Ltd
+     * @param msg:
+     * @return: void
+     * @since: 2020/10/27 14:28
+     * @version: V1.0
+     */
     @Override
     public void handleIOTData(String msg) throws SchedulerException, InterruptedException {
         String[] split = msg.split(";");
@@ -71,7 +80,10 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
         }
         //返回的信息点个数
         int nums = (split.length) / 4;
-        log.info("接收信息length:[{}]数据为:{}", nums,msg);
+        if(log.isDebugEnabled()){
+            log.debug("接收信息length:[{}]数据为:{}", nums,msg);
+        }
+        System.out.println(nums);
         for (int i = 0; i < nums; i++) {
             String dateTime = split[i * 4];
             String meterId = split[i * 4 + 1];
@@ -85,6 +97,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                     Condition condition = alarmDefine.getCondition();
                     List<JSONObject> codeDetail = condition.getInfoCodes();
                     boolean match = codeDetail.stream().allMatch(p -> currentDataCache.hasKey(p.getString("meterId"), p.getString("funcId")));
+                    //报警定义的所有信息点都有采集数值,具备判断条件
                     if (match) {
                         String defineId = AlarmInfoCache.getAlarmDefineId(alarmDefine);
                         String trigger = condition.getTrigger();
@@ -98,8 +111,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                         Expression endExp = AviatorEvaluator.compile(end, true);
                         Boolean triggerResult = (Boolean) triggerExp.execute(paramMap);
                         Boolean endResult = (Boolean) endExp.execute(paramMap);
-                        System.out.println("triggerResult:" + triggerResult);
-                        System.out.println("endResult:" + endResult);
+                        log.info("triggerResult:[{}],endResult:[{}]" , triggerResult,endResult);
                         if (triggerResult && endResult) {
                             log.warn("报警触发条件和报警恢复条件同时满足,请检查,报警定义详情【{}】", alarmDefine.toString());
                         }
@@ -160,35 +172,36 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                     endTime = dateTime;
                 }
                 alarmState.setAlarmEndTime(endTime);
-                alarmState.setAlarmStartTime("");
                 //设置开始恢复时间
                 int uphold = condition.getEndUphold();
                 //超过报警恢复设置的持续时间
-                if (DateUtils.betweenTwoTimeSecond(endTime, dateTime) >= 0) {
+                if (DateUtils.betweenTwoTimeSecond(endTime, dateTime) >= uphold) {
                     log.error("产生一条报警恢复消息");
                     NettyMessage<JSONObject> nettyMessage = new NettyMessage<>();
                     nettyMessage.setOpCode(6);
                     JSONObject result = new JSONObject();
-                    ZktAlarmRecordDO alarmRecordDO = alarmRecordRepository.findById(AlarmInfoCache.getAlarmDefineId(alarmDefine)).orElse(new ZktAlarmRecordDO());
-                    //TODO : 为空处理
-                    result.put("id", alarmRecordDO.getAlarmId());
-                    result.put("state", "2");
-                    result.put("groupCode", groupCode);
-                    result.put("projectId", projectId);
-                    nettyMessage.setContent(Arrays.asList(result));
-                    //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123"}
-                    groupNettyClient.sendMessage(nettyMessage.toString());
-
-                    log.error("产生一条取消报警过期定时任务的消息");
-                    String jobName = AlarmInfoCache.getAlarmDefineId(alarmDefine);
-                    ExpireAlarmMessage em = new ExpireAlarmMessage();
-                    //取消过期消息
-                    em.setType("2");
-                    em.setJobName(jobName);
-                    em.setJobGroupName("expire");
-                    ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
-                    //报警恢复,报警状态重置回默认
-                    alarmState.reset();
+                    Optional<ZktAlarmRecordDO> alarmRecordDO = alarmRecordRepository.findById(AlarmInfoCache.getAlarmDefineId(alarmDefine));
+                    if (alarmRecordDO.isPresent()){
+                        result.put("id", alarmRecordDO.get().getAlarmId());
+                        result.put("state", "2");
+                        result.put("groupCode", groupCode);
+                        result.put("projectId", projectId);
+                        nettyMessage.setContent(Arrays.asList(result));
+                        //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123"}
+                        groupNettyClient.sendMessage(nettyMessage.toString());
+                        log.error("产生一条取消报警过期定时任务的消息");
+                        String jobName = AlarmInfoCache.getAlarmDefineId(alarmDefine);
+                        ExpireAlarmMessage em = new ExpireAlarmMessage();
+                        //取消过期消息
+                        em.setType("2");
+                        em.setJobName(jobName);
+                        em.setJobGroupName("expire");
+                        ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
+                        //报警恢复,报警状态重置回默认
+                        alarmState.reset();
+                    }else {
+                        log.warn("找不到该条报警状态的信息");
+                    }
                 }
             }
             alarmInfoCache.setAlarmState(defineId, alarmState);
@@ -241,6 +254,9 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                 long timeSecond = 0;
                 if (StringUtil.isNotEmpty(alarmState.getAlarmStartTime())) {
                     timeSecond = DateUtils.betweenTwoTimeSecond(alarmState.getAlarmStartTime(), dateTime);
+                }else {
+                    //设置开始报警时间
+                    alarmState.setAlarmStartTime(dateTime);
                 }
                 LocalDateTime expireDateTime = null;
                 Date expireDate = null;
@@ -251,8 +267,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                     expireDateTime = LocalDateTime.of(LocalDate.now(), localTime);
                     expireDate = DateUtils.localDateTime2Date(expireDateTime);
                 }
-                //设置开始报警时间
-                alarmState.setAlarmStartTime(dateTime);
+
                 if (timeSecond >= condition.getTriggerUphold()) {
                     log.error("大于持续时间了,产生一条报警");
                     AlarmRecord alarmRecord = AlarmRecord.builder()
@@ -288,7 +303,8 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                         //过期时间
                         log.error("产生一条定时过期报警消息");
                         String alarmId = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO()).getAlarmId();
-                        ZktAlarmRecordDO alarmRecordDO = ZktAlarmRecordDO.builder().alarmTime(alarmState.getAlarmStartTime())
+                        ZktAlarmRecordDO alarmRecordDO = ZktAlarmRecordDO.builder()
+                                .alarmTime(alarmState.getAlarmStartTime())
                                 .effectEndTime(DateUtils.format(expireDateTime))
                                 .definitionId(defineId)
                                 .itemCode(alarmDefine.getItemCode())
@@ -301,7 +317,8 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                         jobDataMap.put("alarmRecord", alarmRecordDO.toString());
                         jobDataMap.put("refire", "0");
                         jobDataMap.put("expireTime", DateUtils.format(expireDateTime));
-                        String jobName = AlarmInfoCache.getAlarmDefineId(alarmDefine);
+                        jobDataMap.put("defineId", defineId);
+                        String jobName = defineId;
                         ExpireAlarmMessage em = new ExpireAlarmMessage();
                         //过期消息
                         em.setType("1");

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

@@ -1,5 +1,6 @@
 package com.persagy.service.impl;
 
+import cn.hutool.core.date.DateUtil;
 import com.persagy.entity.ExpireAlarmMessage;
 import com.persagy.job.AlarmExpireJob;
 import com.persagy.job.ExpireAlarmQueue;
@@ -31,9 +32,12 @@ public class AlarmQuartzServiceImpl implements AlarmQuartzService {
     @Qualifier("quartzScheduler")
     Scheduler quartzScheduler;
 
-    @PostConstruct
-    public void init() {
+    @Override
+    public void initAlarmMessagehandler() {
         try {
+            System.out.println("--init--");
+            //启动后先休息20秒,这样报警netty  websocket  quartz都启动了
+            Thread.sleep(20000);
             while (true){
                 ExpireAlarmMessage expireAlarmMessage = ExpireAlarmQueue.getExpireAlarmMessageQueue().consume();
                 log.info("剩余过期消息总数:{}",ExpireAlarmQueue.getExpireAlarmMessageQueue().size());
@@ -44,7 +48,7 @@ public class AlarmQuartzServiceImpl implements AlarmQuartzService {
                 }
             }
         } catch (Exception e) {
-            log.info("报警过期消息队列消费失败,失败原因为---->",e.getMessage());
+            log.error("报警过期消息队列消费失败,失败原因为---->",e.getMessage());
         }
     }
     /**
@@ -69,7 +73,7 @@ public class AlarmQuartzServiceImpl implements AlarmQuartzService {
                 .withSchedule(simpleSchedule().withMisfireHandlingInstructionIgnoreMisfires()).build();
         //已经存在的job实例和触发器自动覆盖 job实 例 唯 一标识:
         HashSet<Trigger> triggerSet = new HashSet<Trigger>();
-        Date fireTime = trigger.getFireTimeAfter(startTime);
+        Date fireTime = trigger.getFireTimeAfter(DateUtil.offsetMinute(startTime,-1).toJdkDate());
         if(fireTime != null){
             log.info("执行时间为:【{}】,设置时间为【{}】",DateUtils.formatDate(fireTime),DateUtils.formatDate(startTime));
         }else {

+ 31 - 7
src/main/resources/logback-spring.xml

@@ -8,28 +8,49 @@ configuration 子节点为 appender、logger、root       -->
     <contextName>zk-project-alarm</contextName>
     <!--日志文件所在目录,如果是tomcat,如下写法日志文件会在则为${TOMCAT_HOME}/bin/logs/目录下-->
     <property name="LOG_HOME" value="../logs/zk-project-alarm"/>
+    <!--
+    %p:输出优先级,即DEBUG,INFO,WARN,ERROR,FATAL
+    %r:输出自应用启动到输出该日志讯息所耗费的毫秒数
+    %t:输出产生该日志事件的线程名
+    %f:输出日志讯息所属的类别的类别名
+    %c:输出日志讯息所属的类的全名
+    %d:输出日志时间点的日期或时间,指定格式的方式: %d{yyyy-MM-dd HH:mm:ss}
+    %l:输出日志事件的发生位置,即输出日志讯息的语句在他所在类别的第几行。
+    %m:输出代码中指定的讯息,如log(message)中的message
+    %n:输出一个换行符号
+-->
+    <!--格式化输出:%d表示日期,%thread表示线程名,%-5level:级别从左显示5个字符宽度 %msg:日志消息,%n是换行符-->
+    <property name="pattern" value="%d{yyyy-MM-dd:HH:mm:ss.SSS} [%thread] %-5level  %msg%n"/>
+
     <!--控制台-->
     <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
-        <encoder>
-            <!--格式化输出:%d表示日期,%thread表示线程名,%-5level:级别从左显示5个字符宽度 %logger输出日志的logger名 %msg:日志消息,%n是换行符 -->
-            <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] [%thread] %-5level %logger{36} : %msg%n</pattern>
+        <!-- 字符串System.out(默认)或者System.err -->
+        <target>System.out</target>
+        <!-- 对记录事件进行格式化 -->
+        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+            <pattern>${pattern}</pattern>
             <!--解决乱码问题-->
             <charset>UTF-8</charset>
         </encoder>
     </appender>
     <!--滚动文件-->
     <appender name="infoFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <!-- 被写入的文件名,可以是相对目录,也可以是绝对目录,如果上级目录不存在会自动创建 -->
+        <file>${LOG_HOME}/log.log</file>
         <!-- ThresholdFilter:临界值过滤器,过滤掉 TRACE 和 DEBUG 级别的日志 -->
         <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
             <level>INFO</level>
         </filter>
         <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-            <fileNamePattern>${LOG_HOME}/log.%d{yyyy-MM-dd}.log</fileNamePattern>
+            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
+                <maxFileSize>100MB</maxFileSize>
+            </timeBasedFileNamingAndTriggeringPolicy>
+            <fileNamePattern>${LOG_HOME}/log.%d{yyyy-MM-dd}.log.%i.gz</fileNamePattern>
             <maxHistory>30</maxHistory><!--保存最近30天的日志-->
         </rollingPolicy>
         <encoder>
             <charset>UTF-8</charset>
-            <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] [%thread] %-5level %logger{36} : %msg%n</pattern>
+            <pattern>${pattern}</pattern>
         </encoder>
     </appender>
     <!--滚动文件-->
@@ -39,7 +60,10 @@ configuration 子节点为 appender、logger、root       -->
             <level>error</level>
         </filter>
         <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-            <fileNamePattern>${LOG_HOME}/error.%d{yyyy-MM-dd}.log</fileNamePattern>
+            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
+                <maxFileSize>100MB</maxFileSize>
+            </timeBasedFileNamingAndTriggeringPolicy>
+            <fileNamePattern>${LOG_HOME}/error.%d{yyyy-MM-dd}.log.%i.gz</fileNamePattern>
             <maxHistory>30</maxHistory><!--保存最近30天的日志-->
         </rollingPolicy>
         <encoder>
@@ -63,7 +87,7 @@ configuration 子节点为 appender、logger、root       -->
     <root level="INFO">
         <appender-ref ref="stdout"/>
         <appender-ref ref="infoFile"/>
-        <appender-ref ref="errorFile"/>
+        <!--<appender-ref ref="errorFile"/>-->
         <!--<appender-ref ref="logstash"/>-->
     </root>