瀏覽代碼

1.内存获取为空时候先从数据库同步一下,避免重启服务产生重复报警
2.IOT多线程判断时候增加同步控制,同一个报警定义的串行,不同报警定义并行

lgy 4 年之前
父節點
當前提交
095fa90e27

+ 6 - 0
src/main/java/com/persagy/entity/ZktAlarmRecordDO.java

@@ -110,6 +110,12 @@ public class ZktAlarmRecordDO implements Serializable {
      */
     @ApiModelProperty(value = "报警自动过期时间")
     private Date endTime;
+    /**
+     * 报警名称
+     */
+    @Column(name = "project_id")
+    @ApiModelProperty(value = "项目ID")
+    private String projectId;
 
     @Override
     public String toString() {

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

@@ -35,7 +35,7 @@ public class SpringSchedule {
     }
 
 
-    @Scheduled(initialDelay=1000, fixedRate=600000)
+    @Scheduled(initialDelay=1000, fixedDelay=600000)
     public void connectAnalizeCron() {
         if(Objects.isNull(GroupNettyClient.channelGroup)) {
             log.warn("NettyClient is not init");

+ 272 - 200
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -89,60 +89,71 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
             String dateTime = split[i * 4];
             String meterId = split[i * 4 + 1];
             String funcId = split[i * 4 + 2];
-            if("1001".equals(meterId) && "102".equals(funcId)){
-                System.out.println("-----"+split[i * 4 + 3]);
-            }
             double value = Double.parseDouble(split[i * 4 + 3]);
 
             if (alarmInfoCache.hasKey(meterId, funcId)) {
-                currentDataCache.putCurrentData(meterId, funcId, value);
-                List<AlarmDefine> alarmDefines = alarmInfoCache.getAlarmDefinitionIdByMeterFuncId(meterId, funcId);
-                for (AlarmDefine alarmDefine : alarmDefines) {
-                    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();
-                        String end = condition.getEnd();
-                        HashMap paramMap = new HashMap<>();
-                        for (int j = 0; j < codeDetail.size(); j++) {
-                            JSONObject code = codeDetail.get(j);
-                            paramMap.put(code.getString("infoCode"), currentDataCache.getCurrentData(code.getString("meterId"), code.getString("funcId")));
-                        }
-                        Expression triggerExp = AviatorEvaluator.compile(trigger, true);
-                        Expression endExp = AviatorEvaluator.compile(end, true);
-                        Boolean triggerResult = (Boolean) triggerExp.execute(paramMap);
-                        Boolean endResult = (Boolean) endExp.execute(paramMap);
-                        log.info("triggerResult:[{}],endResult:[{}]" , triggerResult,endResult);
-                        if (triggerResult && endResult) {
-                            log.warn("报警触发条件和报警恢复条件同时满足,请检查,报警定义详情【{}】", alarmDefine.toString());
-                        }
-                        AlarmState alarmState = alarmInfoCache.getAlarmState(defineId);
-                        if (Objects.isNull(alarmState)) {
-                            alarmState = new AlarmState(defineId);
-                        }
-                        //报警产生值满足(这里的满足不考虑报警持续时间)
-                        if (triggerResult) {
-                            log.info("有一条满足报警条件");
-                            log.info("--"+alarmDefine.toString());
-                            log.info("--"+JSONObject.toJSONString(paramMap));
-                            //报警的时候不考虑报警恢复,因为同时报警和报警恢复是不应该出现的
-                            handlerNowDataAlarm(alarmDefine, alarmState, dateTime, condition, defineId,paramMap);
-                        } else {
-                            log.info("不满足报警条件");
-                            log.info("--"+alarmDefine.toString());
-                            log.info("--"+JSONObject.toJSONString(paramMap));
-                            //当前数据正常
-                            handlerNowDataNormal(alarmDefine, dateTime, condition, defineId, endResult, alarmState,paramMap);
+                validIotData(dateTime, meterId, funcId, value);
+            }
+
+        }
+    }
+
+    private void validIotData(String dateTime, String meterId, String funcId, double value) throws SchedulerException, InterruptedException {
+        currentDataCache.putCurrentData(meterId, funcId, value);
+        List<AlarmDefine> alarmDefines = alarmInfoCache.getAlarmDefinitionIdByMeterFuncId(meterId, funcId);
+        for (AlarmDefine alarmDefine : alarmDefines) {
+            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();
+                String end = condition.getEnd();
+                HashMap paramMap = new HashMap<>();
+                for (int j = 0; j < codeDetail.size(); j++) {
+                    JSONObject code = codeDetail.get(j);
+                    paramMap.put(code.getString("infoCode"), currentDataCache.getCurrentData(code.getString("meterId"), code.getString("funcId")));
+                }
+                Expression triggerExp = AviatorEvaluator.compile(trigger, true);
+                Expression endExp = AviatorEvaluator.compile(end, true);
+                Boolean triggerResult = (Boolean) triggerExp.execute(paramMap);
+                Boolean endResult = (Boolean) endExp.execute(paramMap);
+                log.info("triggerResult:[{}],endResult:[{}]" , triggerResult,endResult);
+                if (triggerResult && endResult) {
+                    log.warn("报警触发条件和报警恢复条件同时满足,请检查,报警定义详情【{}】", alarmDefine.toString());
+                }
+                AlarmState alarmState = alarmInfoCache.getAlarmState(defineId);
+                if (Objects.isNull(alarmState)) {
+                    //默认正常报警状态
+                    alarmState = new AlarmState(defineId);
+                    Optional<ZktAlarmRecordDO> recordOptional = alarmRecordRepository.findById(defineId);
+                    if(recordOptional.isPresent()){
+                        ZktAlarmRecordDO alarmRecordDO = recordOptional.get();
+                        //数据库报警状态:1-未处理
+                        if("1".equals(alarmRecordDO.getState())){
+                            alarmState.setState(AlarmState.State.NOT_DEAL.getType());
+                            alarmState.setAlarmStartTime(alarmRecordDO.getAlarmTime());
                         }
-                    } else {
-                        log.warn("部分信息点没有数值:[{}]", codeDetail);
                     }
                 }
+                //报警产生值满足(这里的满足不考虑报警持续时间)
+                if (triggerResult) {
+                    log.info("有一条满足报警条件");
+                    log.info("--"+alarmDefine.toString());
+                    log.info("--"+JSONObject.toJSONString(paramMap));
+                    //报警的时候不考虑报警恢复,因为同时报警和报警恢复是不应该出现的
+                    handlerNowDataAlarm(alarmDefine, alarmState, dateTime, condition, defineId,paramMap,meterId,funcId,value);
+                } else {
+                    log.info("不满足报警条件");
+                    log.info("--"+alarmDefine.toString());
+                    log.info("--"+JSONObject.toJSONString(paramMap));
+                    //当前数据正常
+                    handlerNowDataNormal(alarmDefine, dateTime, condition, defineId, endResult, alarmState,paramMap,meterId,funcId,value);
+                }
+            } else {
+                log.warn("部分信息点没有数值:[{}]", codeDetail);
             }
-
         }
     }
 
@@ -154,6 +165,9 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
      * @param endResult :报警恢复监测结果
      * @param alarmState :        报警状态时间实体
      * @param paramMap   报警恢复触发值
+     * @param meterId:  表号
+     * @param funcId: 功能号
+     * @param value:  数值
      * @description:当前数据正常判断逻辑
      * @exception:
      * @author: LuoGuangyi
@@ -162,87 +176,105 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
      * @since: 2020/10/21 16:46
      * @version: V1.0
      */
-    private void handlerNowDataNormal(AlarmDefine alarmDefine, String dateTime, Condition condition, String defineId, Boolean endResult, AlarmState alarmState, HashMap paramMap) throws InterruptedException {
+    private void handlerNowDataNormal(AlarmDefine alarmDefine, String dateTime, Condition condition, String defineId, Boolean endResult, AlarmState alarmState, HashMap paramMap, String meterId, String funcId, double value) throws InterruptedException, SchedulerException {
         //当前数据正常,报警状态为正常:清空之前的报警计时,重置回默认状态
         if (AlarmState.State.NORMAL.getType().equals(alarmState.getState())) {
             alarmState = new AlarmState(defineId);
             alarmInfoCache.setAlarmState(defineId, alarmState);
         } else if (AlarmState.State.NOT_DEAL.getType().equals(alarmState.getState())) {
-            //报警状态异常时候
-            alarmState.setLatestDataNormalstate(true);
-            //报警恢复条件满足,判断是否满足报警恢复持续时间
-            if (endResult) {
-                String endTime = alarmState.getAlarmEndTime();
-                if (StringUtil.isEmpty(endTime)) {
-                    endTime = dateTime;
+            synchronized (defineId.intern()){
+                AlarmState latestAlarmState = alarmInfoCache.getAlarmState(defineId);
+                if (ObjectUtil.isNull(latestAlarmState)) {
+                    //默认正常报警状态
+                    latestAlarmState = new AlarmState(defineId);
                 }
-                alarmState.setAlarmEndTime(endTime);
-                //设置开始恢复时间
-                int uphold = condition.getEndUphold();
-                //超过报警恢复设置的持续时间
-                if (DateUtils.betweenTwoTimeSecond(endTime, dateTime) >= uphold) {
-                    log.error("产生一条报警恢复消息[{}]>[{}]",DateUtils.betweenTwoTimeSecond(endTime, dateTime),uphold);
-                    NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(6,projectId);
-                    ZktAlarmRecordDO alarmRecordDO = alarmRecordRepository.findById(AlarmInfoCache.getAlarmDefineId(alarmDefine)).orElse(new ZktAlarmRecordDO());
-                    alarmRecordDO.setDefinitionId(defineId);
-                    alarmRecordDO.setObjId(alarmDefine.getObjId());
-                    alarmRecordDO.setItemCode(alarmDefine.getItemCode());
-                    alarmRecordDO.setTargetId(alarmDefine.getTargetId());
-                    alarmRecordDO.setState("2");
-                    alarmRecordDO.setEndTime(DateUtils.parseDate(dateTime));
-                    alarmRecordDO.setEndInfo(JSONObject.toJSONString(paramMap));
-                    //更新报警状态
-                    alarmRecordRepository.save(alarmRecordDO);
-                    String alarmId = alarmRecordDO.getAlarmId();
-                    //报警恢复参数
-                    AlarmRecord alarmResumeRecord = AlarmRecord.builder()
-                            .state(2)
-                            .groupCode(groupCode)
-                            .projectId(projectId)
-                            .endTime(DateUtils.parseDate(dateTime))
-                            .endInfo(JSONObject.toJSONString(paramMap))
-                            .build();
-                    //如果有报警ID,直接报警恢复
-                    if(StringUtils.isNotEmpty(alarmId)) {
-                        alarmResumeRecord.setId(alarmId);
-                        nettyMessage.setContent(Arrays.asList(alarmResumeRecord));
-                        //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123","endTime":"","endInfo":""}
-                        groupNettyClient.sendMessage(nettyMessage.toString());
-                    }else {
-                        //如果没有报警ID,定时任务再次测试
-                        JobDataMap jobDataMap = new JobDataMap();
-                        jobDataMap.put("alarmRecord", alarmRecordDO.toString());
-                        jobDataMap.put("refire", "0");
-                        jobDataMap.put("endTime", dateTime);
-                        jobDataMap.put("endInfo", JSONObject.toJSONString(paramMap));
-                        jobDataMap.put("defineId", defineId);
-                        //恢复
-                        jobDataMap.put("state", "2");
-                        log.info(JSONObject.toJSONString(jobDataMap));
-                        String jobName = defineId;
-                        ExpireAlarmMessage em = new ExpireAlarmMessage();
-                        //过期消息
-                        em.setType("1");
-                        em.setStartTime(DateUtil.offsetMinute(new Date(),3).toJdkDate());
-                        em.setJobDataMap(jobDataMap);
-                        em.setJobName(jobName);
-                        em.setJobGroupName("resume");
-                        ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
-                    }
+                //上锁前报警状态和上锁后状态一样,可以直接报警生成
+                if(alarmState.getState().equals(latestAlarmState.getState())){
+                    handlerNormalWithLock(alarmDefine, dateTime, condition, defineId, endResult, alarmState, paramMap);
+                }else {
+                    //上锁前后状态不一致了,重新进行报警计算
+                    log.warn("报警定义的报警状态加锁前后发生变化,加锁前[{}],加锁后[{}]",alarmState,latestAlarmState);
+                    validIotData(dateTime,meterId,funcId,value);
+                }
+            }
+        }
+    }
 
-                    String jobName = AlarmInfoCache.getAlarmDefineId(alarmDefine);
+    private void handlerNormalWithLock(AlarmDefine alarmDefine, String dateTime, Condition condition, String defineId, Boolean endResult, AlarmState alarmState, HashMap paramMap) throws InterruptedException {
+        //报警状态异常时候
+        alarmState.setLatestDataNormalstate(true);
+        //报警恢复条件满足,判断是否满足报警恢复持续时间
+        if (endResult) {
+            String endTime = alarmState.getAlarmEndTime();
+            if (StringUtil.isEmpty(endTime)) {
+                endTime = dateTime;
+            }
+            alarmState.setAlarmEndTime(endTime);
+            //设置开始恢复时间
+            int uphold = condition.getEndUphold();
+            //超过报警恢复设置的持续时间
+            if (DateUtils.betweenTwoTimeSecond(endTime, dateTime) >= uphold) {
+                log.error("产生一条报警恢复消息[{}]>[{}]",DateUtils.betweenTwoTimeSecond(endTime, dateTime),uphold);
+                NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(6,projectId);
+                ZktAlarmRecordDO alarmRecordDO = alarmRecordRepository.findById(AlarmInfoCache.getAlarmDefineId(alarmDefine)).orElse(new ZktAlarmRecordDO());
+                alarmRecordDO.setDefinitionId(defineId);
+                alarmRecordDO.setObjId(alarmDefine.getObjId());
+                alarmRecordDO.setItemCode(alarmDefine.getItemCode());
+                alarmRecordDO.setTargetId(alarmDefine.getTargetId());
+                alarmRecordDO.setState("2");
+                alarmRecordDO.setEndTime(DateUtils.parseDate(dateTime));
+                alarmRecordDO.setEndInfo(JSONObject.toJSONString(paramMap));
+                //更新报警状态
+                alarmRecordRepository.save(alarmRecordDO);
+                String alarmId = alarmRecordDO.getAlarmId();
+                //报警恢复参数
+                AlarmRecord alarmResumeRecord = AlarmRecord.builder()
+                        .state(2)
+                        .groupCode(groupCode)
+                        .projectId(projectId)
+                        .endTime(DateUtils.parseDate(dateTime))
+                        .endInfo(JSONObject.toJSONString(paramMap))
+                        .build();
+                //如果有报警ID,直接报警恢复
+                if(StringUtils.isNotEmpty(alarmId)) {
+                    alarmResumeRecord.setId(alarmId);
+                    nettyMessage.setContent(Arrays.asList(alarmResumeRecord));
+                    //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123","endTime":"","endInfo":""}
+                    groupNettyClient.sendMessage(nettyMessage.toString());
+                }else {
+                    //如果没有报警ID,定时任务再次测试
+                    JobDataMap jobDataMap = new JobDataMap();
+                    jobDataMap.put("alarmRecord", alarmRecordDO.toString());
+                    jobDataMap.put("refire", "0");
+                    jobDataMap.put("endTime", dateTime);
+                    jobDataMap.put("endInfo", JSONObject.toJSONString(paramMap));
+                    jobDataMap.put("defineId", defineId);
+                    //恢复
+                    jobDataMap.put("state", "2");
+                    log.info(JSONObject.toJSONString(jobDataMap));
+                    String jobName = defineId;
                     ExpireAlarmMessage em = new ExpireAlarmMessage();
-                    //取消过期消息
-                    em.setType("2");
+                    //过期消息
+                    em.setType("1");
+                    em.setStartTime(DateUtil.offsetMinute(new Date(),3).toJdkDate());
+                    em.setJobDataMap(jobDataMap);
                     em.setJobName(jobName);
-                    em.setJobGroupName("expire");
+                    em.setJobGroupName("resume");
                     ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
-                    //报警恢复,报警状态重置回默认
-                    alarmState.reset();
                 }
+
+                String jobName = AlarmInfoCache.getAlarmDefineId(alarmDefine);
+                ExpireAlarmMessage em = new ExpireAlarmMessage();
+                //取消过期消息
+                em.setType("2");
+                em.setJobName(jobName);
+                em.setJobGroupName("expire");
+                ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
+                //报警恢复,报警状态重置回默认
+                alarmState.reset();
             }
-            alarmInfoCache.setAlarmState(defineId, alarmState);
         }
+        alarmInfoCache.setAlarmState(defineId, alarmState);
     }
 
     /**
@@ -251,6 +283,9 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
      * @param condition   :报警定义条件
      * @param defineId    :报警定义ID
      * @param paramMap    报警触发值
+     * @param meterId    表号
+     * @param funcId    功能号
+     * @param value     IOT采集数值
      * @description:处理当前值报警的情况
      * @exception:
      * @author: LuoGuangyi
@@ -259,7 +294,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
      * @since: 2020/10/20 22:31
      * @version: V1.0
      */
-    private void handlerNowDataAlarm(AlarmDefine alarmDefine, AlarmState alarmState, String dateTime, Condition condition, String defineId, HashMap paramMap) throws SchedulerException, InterruptedException {
+    private void handlerNowDataAlarm(AlarmDefine alarmDefine, AlarmState alarmState, String dateTime, Condition condition, String defineId, HashMap paramMap, String meterId, String funcId, double value) throws SchedulerException, InterruptedException {
         alarmState.setLatestDataNormalstate(false);
         JSONObject effectTime = condition.getEffectTime();
         //无生效时间设置,则任何时间生效
@@ -287,91 +322,20 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
             //之前是报警状态,现在还是报警,报警值没变化
             //之前是是正常时候
             if (AlarmState.State.NORMAL.getType().equals(alarmState.getState())) {
-                long timeSecond = 0;
-                if (StringUtil.isNotEmpty(alarmState.getAlarmStartTime())) {
-                    timeSecond = DateUtils.betweenTwoTimeSecond(alarmState.getAlarmStartTime(), dateTime);
-                }else {
-                    //设置开始报警时间
-                    alarmState.setAlarmStartTime(dateTime);
-                }
-                LocalDateTime expireDateTime = null;
-                Date expireDate = null;
-                if (hasExpire && "period".equals(effectTime.getString("type"))) {
-                    //过期时间
-                    String expireTime = period.getString("endTime");
-                    LocalTime localTime = LocalTime.parse(expireTime, DateTimeFormatter.ofPattern(DateUtils.sdfTimeNotDate));
-                    expireDateTime = LocalDateTime.of(LocalDate.now(), localTime);
-                    expireDate = DateUtils.localDateTime2Date(expireDateTime);
-                }
-
-                if (timeSecond >= condition.getTriggerUphold()) {
-                    log.warn("大于持续时间了,产生一条报警[{}]>[{}]",timeSecond,condition.getTriggerUphold());
-                    AlarmRecord alarmRecord = AlarmRecord.builder()
-                            .category(alarmDefine.getCategory())
-                            .concern(alarmDefine.getConcern())
-                            .level(alarmDefine.getLevel())
-                            .projectId(projectId)
-                            .state(1)
-                            .triggerTime(DateUtils.parseDate(dateTime))
-                            .remark(alarmDefine.getRemark())
-                            .triggerInfo(JSONObject.toJSONString(paramMap))
-                            .condition(condition.toString())
-                            .effectEndTime(expireDate)
-                            .groupCode(groupCode)
-                            .itemCode(alarmDefine.getItemCode())
-                            .objId(alarmDefine.getObjId())
-                            .targetId(alarmDefine.getTargetId())
-                            .classCode(alarmDefine.getClassCode())
-                            .createUser(systemId)
-                            .build();
-                    NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(5,projectId);
-                    nettyMessage.setContent(Arrays.asList(alarmRecord));
-                    //推送一条报警记录给远端
-                    groupNettyClient.sendMessage(nettyMessage.toString());
-                    ZktAlarmRecordDO zktAlarmRecordDO = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO());
-                    zktAlarmRecordDO.setDefinitionId(defineId);
-                    zktAlarmRecordDO.setObjId(alarmDefine.getObjId());
-                    zktAlarmRecordDO.setItemCode(alarmDefine.getItemCode());
-                    zktAlarmRecordDO.setTargetId(alarmDefine.getTargetId());
-                    zktAlarmRecordDO.setState("1");
-                    alarmRecordRepository.save(zktAlarmRecordDO);
-                    alarmState.setState(AlarmState.State.NOT_DEAL.getType());
-                    //有过期时间,生成报警过期消息
-                    if (hasExpire && "period".equals(effectTime.getString("type"))) {
-                        //过期时间
-                        log.error("产生一条定时过期报警消息");
-                        String alarmId = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO()).getAlarmId();
-                        ZktAlarmRecordDO alarmRecordDO = ZktAlarmRecordDO.builder()
-                                .alarmTime(alarmState.getAlarmStartTime())
-                                .effectEndTime(DateUtils.format(expireDateTime))
-                                .definitionId(defineId)
-                                .itemCode(alarmDefine.getItemCode())
-                                .name(alarmDefine.getName())
-                                .objId(alarmDefine.getObjId())
-                                .targetId(alarmDefine.getTargetId())
-                                .state("3")  //要变成已过期
-                                .alarmId(alarmId)
-                                .build();
-                        JobDataMap jobDataMap = new JobDataMap();
-                        jobDataMap.put("alarmRecord", alarmRecordDO.toString());
-                        jobDataMap.put("refire", "0");
-                        jobDataMap.put("expireTime", DateUtils.format(expireDateTime));
-                        jobDataMap.put("defineId", defineId);
-                        //过期
-                        jobDataMap.put("state", "3");
-                        String jobName = defineId;
-                        ExpireAlarmMessage em = new ExpireAlarmMessage();
-                        //过期消息
-                        em.setType("1");
-                        em.setStartTime(DateUtils.localDateTime2Date(expireDateTime));
-                        em.setJobDataMap(jobDataMap);
-                        em.setJobName(jobName);
-                        em.setJobGroupName("expire");
-                        ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
-                        //alarmQuartzService.addExpireJob(DateUtils.localDateTime2Date(expireDateTime), jobName, "expire", jobDataMap);
+                synchronized (defineId.intern()) {
+                    AlarmState latestAlarmState = alarmInfoCache.getAlarmState(defineId);
+                    if (ObjectUtil.isNull(latestAlarmState)) {
+                        //默认正常报警状态
+                        latestAlarmState = new AlarmState(defineId);
+                    }
+                    //上锁前报警状态和上锁后状态一样,可以直接报警生成
+                    if(alarmState.getState().equals(latestAlarmState.getState())){
+                        handlerAlarmWithLock(alarmDefine, alarmState, dateTime, condition, defineId, paramMap, effectTime, hasExpire, period);
+                    }else {
+                        //上锁前后状态不一致了,重新进行报警计算
+                        log.warn("报警定义的报警状态加锁前后发生变化,加锁前[{}],加锁后[{}]",alarmState,latestAlarmState);
+                        validIotData(dateTime,meterId,funcId,value);
                     }
-                } else {
-                    alarmState.setState(AlarmState.State.NORMAL.getType());
                 }
             }
         } else {
@@ -381,4 +345,112 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
         }
         alarmInfoCache.setAlarmState(defineId, alarmState);
     }
+    /**
+     * @description: 生成报警记录和报警恢复消息(对报警定义加锁)
+     * @author: LuoGuangyi
+     * @createTime: 2020/12/28 18:21
+     * @param alarmDefine:
+     * @param alarmState:
+     * @param dateTime:
+     * @param condition:
+     * @param defineId:
+     * @param paramMap:
+     * @param effectTime:
+     * @param hasExpire:
+     * @param period:
+     * @return: void
+     * @expression
+     */
+    private void handlerAlarmWithLock(AlarmDefine alarmDefine, AlarmState alarmState, String dateTime, Condition condition, String defineId, HashMap paramMap, JSONObject effectTime, boolean hasExpire, JSONObject period) throws InterruptedException {
+        long timeSecond = 0;
+        if (StringUtil.isNotEmpty(alarmState.getAlarmStartTime())) {
+            timeSecond = DateUtils.betweenTwoTimeSecond(alarmState.getAlarmStartTime(), dateTime);
+        }else {
+            //设置开始报警时间
+            alarmState.setAlarmStartTime(dateTime);
+        }
+        LocalDateTime expireDateTime = null;
+        Date expireDate = null;
+        if (hasExpire && "period".equals(effectTime.getString("type"))) {
+            //过期时间
+            String expireTime = period.getString("endTime");
+            LocalTime localTime = LocalTime.parse(expireTime, DateTimeFormatter.ofPattern(DateUtils.sdfTimeNotDate));
+            expireDateTime = LocalDateTime.of(LocalDate.now(), localTime);
+            expireDate = DateUtils.localDateTime2Date(expireDateTime);
+        }
+
+        if (timeSecond >= condition.getTriggerUphold()) {
+            log.warn("大于持续时间了,产生一条报警[{}]>[{}]",timeSecond, condition.getTriggerUphold());
+            AlarmRecord alarmRecord = AlarmRecord.builder()
+                    .category(alarmDefine.getCategory())
+                    .concern(alarmDefine.getConcern())
+                    .level(alarmDefine.getLevel())
+                    .projectId(projectId)
+                    .state(1)
+                    //报警时间为第一次满足报警条件时候的时间
+                    .triggerTime(DateUtils.parseDate(alarmState.getAlarmStartTime()))
+                    .remark(alarmDefine.getRemark())
+                    .triggerInfo(JSONObject.toJSONString(paramMap))
+                    .condition(condition.toString())
+                    .effectEndTime(expireDate)
+                    .groupCode(groupCode)
+                    .itemCode(alarmDefine.getItemCode())
+                    .objId(alarmDefine.getObjId())
+                    .targetId(alarmDefine.getTargetId())
+                    .classCode(alarmDefine.getClassCode())
+                    .createUser(systemId)
+                    .build();
+            NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(5,projectId);
+            nettyMessage.setContent(Arrays.asList(alarmRecord));
+            //推送一条报警记录给远端
+            groupNettyClient.sendMessage(nettyMessage.toString());
+            ZktAlarmRecordDO zktAlarmRecordDO = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO());
+            zktAlarmRecordDO.setDefinitionId(defineId);
+            zktAlarmRecordDO.setObjId(alarmDefine.getObjId());
+            zktAlarmRecordDO.setItemCode(alarmDefine.getItemCode());
+            zktAlarmRecordDO.setTargetId(alarmDefine.getTargetId());
+            zktAlarmRecordDO.setState("1");
+            zktAlarmRecordDO.setAlarmTime(alarmState.getAlarmStartTime());
+            zktAlarmRecordDO.setProjectId(alarmDefine.getProjectId());
+            alarmRecordRepository.save(zktAlarmRecordDO);
+            alarmState.setState(AlarmState.State.NOT_DEAL.getType());
+            //有过期时间,生成报警过期消息
+            if (hasExpire && "period".equals(effectTime.getString("type"))) {
+                //过期时间
+                log.error("产生一条定时过期报警消息");
+                String alarmId = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO()).getAlarmId();
+                ZktAlarmRecordDO alarmRecordDO = ZktAlarmRecordDO.builder()
+                        .alarmTime(alarmState.getAlarmStartTime())
+                        .effectEndTime(DateUtils.format(expireDateTime))
+                        .definitionId(defineId)
+                        .itemCode(alarmDefine.getItemCode())
+                        .name(alarmDefine.getName())
+                        .objId(alarmDefine.getObjId())
+                        .targetId(alarmDefine.getTargetId())
+                        .projectId(alarmDefine.getProjectId())
+                        .state("3")  //要变成已过期
+                        .alarmId(alarmId)
+                        .build();
+                JobDataMap jobDataMap = new JobDataMap();
+                jobDataMap.put("alarmRecord", alarmRecordDO.toString());
+                jobDataMap.put("refire", "0");
+                jobDataMap.put("expireTime", DateUtils.format(expireDateTime));
+                jobDataMap.put("defineId", defineId);
+                //过期
+                jobDataMap.put("state", "3");
+                String jobName = defineId;
+                ExpireAlarmMessage em = new ExpireAlarmMessage();
+                //过期消息
+                em.setType("1");
+                em.setStartTime(DateUtils.localDateTime2Date(expireDateTime));
+                em.setJobDataMap(jobDataMap);
+                em.setJobName(jobName);
+                em.setJobGroupName("expire");
+                ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
+                //alarmQuartzService.addExpireJob(DateUtils.localDateTime2Date(expireDateTime), jobName, "expire", jobDataMap);
+            }
+        } else {
+            alarmState.setState(AlarmState.State.NORMAL.getType());
+        }
+    }
 }