123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444 |
- package com.persagy.service.impl;
- import static org.apache.commons.lang3.math.NumberUtils.INTEGER_ONE;
- import java.time.LocalDate;
- import java.time.LocalDateTime;
- import java.time.LocalTime;
- import java.time.format.DateTimeFormatter;
- import java.util.Arrays;
- import java.util.Date;
- import java.util.HashMap;
- import java.util.List;
- import java.util.Objects;
- import java.util.Optional;
- import org.apache.commons.lang3.StringUtils;
- import org.quartz.JobDataMap;
- import org.quartz.SchedulerException;
- import org.springframework.beans.factory.annotation.Autowired;
- import org.springframework.stereotype.Service;
- import com.alibaba.fastjson.JSONObject;
- import com.googlecode.aviator.AviatorEvaluator;
- import com.googlecode.aviator.Expression;
- import com.persagy.cache.AlarmInfoCache;
- import com.persagy.cache.CurrentDataCache;
- import com.persagy.constant.CommonConst;
- import com.persagy.entity.AlarmDefine;
- import com.persagy.entity.AlarmRecord;
- import com.persagy.entity.AlarmState;
- import com.persagy.entity.Condition;
- import com.persagy.entity.ExpireAlarmMessage;
- import com.persagy.entity.NettyMessage;
- import com.persagy.entity.ZktAlarmRecordDO;
- import com.persagy.job.ExpireAlarmQueue;
- import com.persagy.netty.client.NettyClient;
- import com.persagy.repository.AlarmRecordRepository;
- import com.persagy.service.AlarmHandleService;
- import com.persagy.service.AlarmQuartzService;
- import com.persagy.utils.DateUtils;
- import com.persagy.utils.StringUtil;
- import cn.hutool.core.date.DateUtil;
- import cn.hutool.core.util.ObjectUtil;
- import lombok.extern.slf4j.Slf4j;
- /**
- * @description:报警处理实现类:包含报警产生和报警恢复
- * @author:LuoGuanhyi
- * @company:PersagyTechnologyCo.,Ltd
- * @since:2020/10/15 001511:14
- * @version:V1.0
- **/
- @Service
- @Slf4j
- public class AlarmHandleServiceImpl implements AlarmHandleService {
- @Autowired
- AlarmInfoCache alarmInfoCache;
- @Autowired
- CurrentDataCache currentDataCache;
- @Autowired
- NettyClient nettyClient;
- @Autowired
- AlarmQuartzService alarmQuartzService;
- @Autowired
- AlarmRecordRepository alarmRecordRepository;
- /**
- * @param msg:
- * @description:处理iot采集数据
- * @exception:
- * @author: LuoGuangyi
- * @company: Persagy Technology Co.,Ltd
- * @return: void
- * @since: 2020/10/27 14:28
- * @version: V1.0
- */
- @Override
- public void handleIOTData(String msg) throws SchedulerException, InterruptedException {
- JSONObject parseObject = JSONObject.parseObject(msg);
- msg = parseObject.getString("data");
- String[] split = msg.split(";");
- if (split.length % 4 != 0) {
- return;
- }
- //返回的信息点个数
- int nums = (split.length) / 4;
- for (int i = 0; i < nums; i++) {
- String dateTime = split[i * 4];
- String meterId = split[i * 4 + 1];
- String funcId = split[i * 4 + 2];
- double value = Double.parseDouble(split[i * 4 + 3]);
- if (alarmInfoCache.hasKey(meterId, funcId)) {
- validIotData(dateTime, meterId, funcId, value);
- }
- }
- }
- private void validIotData(String dateTime, String meterId, String funcId, double value) throws SchedulerException, InterruptedException {
- List<AlarmDefine> alarmDefines = alarmInfoCache.getAlarmDefinitionIdByMeterFuncId(meterId, funcId);
- for (AlarmDefine alarmDefine : alarmDefines) {
- String defineId = AlarmInfoCache.getAlarmDefineId(alarmDefine);
- synchronized (defineId.intern()) {
- currentDataCache.putCurrentData(meterId, funcId, value);
- 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) {
- if(condition.getRealRestraintTime()!=null) {
- if (condition.getRealRestraintTime().compareTo(LocalDateTime.now()) > 0) {
- continue;
- }
- }
- 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());
- }
- }
- }
- //报警产生值满足(这里的满足不考虑报警持续时间)
- if (triggerResult) {
- log.info("有一条满足报警条件{},-----{}----{}", defineId, paramMap, alarmDefine.getCondition());
- if (alarmDefine.getOpen() == 0) {
- log.info("报警定义ID为[{}]已经屏蔽", defineId);
- continue;
- }
- if (AlarmInfoCache.isolationSystemList.contains(alarmDefine.getSystemCode())) {
- log.info("报警定义ID为[{}]的系统[{}]已经隔离,不产生报警", defineId, alarmDefine.getSystemCode());
- continue;
- }
- //报警的时候不考虑报警恢复,因为同时报警和报警恢复是不应该出现的
- handlerNowDataAlarm(alarmDefine, alarmState, dateTime, condition, defineId, paramMap, meterId, funcId, value);
- } else {
- log.info("不满足报警条件{},{}----{}", defineId, paramMap, alarmDefine.getCondition());
- //当前数据正常
- handlerNowDataNormal(alarmDefine, dateTime, condition, defineId, endResult, alarmState, paramMap, meterId, funcId, value);
- }
- } else {
- log.warn("部分信息点没有数值:[{}]", codeDetail);
- }
- }
- }
- }
- /**
- * @param alarmDefine : 报警定义
- * @param dateTime :IOT采集时间
- * @param condition :报警条件
- * @param defineId :报警定义ID
- * @param endResult :报警恢复监测结果
- * @param alarmState : 报警状态时间实体
- * @param paramMap 报警恢复触发值
- * @param meterId: 表号
- * @param funcId: 功能号
- * @param value: 数值
- * @description:当前数据正常判断逻辑
- * @exception:
- * @author: LuoGuangyi
- * @company: Persagy Technology Co.,Ltd
- * @return: void
- * @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, 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())) {
- handlerNormalWithLock(alarmDefine, dateTime, condition, defineId, endResult, alarmState, paramMap);
- }
- }
- 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.info("产生一条报警恢复消息[{}]>[{}]", DateUtils.betweenTwoTimeSecond(endTime, dateTime), uphold);
- NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(6);
- ZktAlarmRecordDO alarmRecordDO = alarmRecordRepository.findById(AlarmInfoCache.getAlarmDefineId(alarmDefine)).orElse(new ZktAlarmRecordDO());
- alarmRecordDO.setDefinitionId(defineId);
- alarmRecordDO.setObjId(alarmDefine.getObjId());
- alarmRecordDO.setItemCode(alarmDefine.getItemCode());
- alarmRecordDO.setItemId(alarmDefine.getItemId());
- 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(CommonConst.groupCode)
- .projectId(CommonConst.projectId)
- .endTime(DateUtils.parse(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":""}
- nettyClient.sendMessage(nettyMessage);
- } 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);
- }
- 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);
- }
- /**
- * @param alarmDefine 报警定义
- * @param dateTime :IOT数据采集时间
- * @param condition :报警定义条件
- * @param defineId :报警定义ID
- * @param paramMap 报警触发值
- * @param meterId 表号
- * @param funcId 功能号
- * @param value IOT采集数值
- * @description:处理当前值报警的情况
- * @exception:
- * @author: LuoGuangyi
- * @company: Persagy Technology Co.,Ltd
- * @return: void
- * @since: 2020/10/20 22:31
- * @version: V1.0
- */
- 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();
- //无生效时间设置,则任何时间生效
- boolean inEffectTime = true;
- //没有设置生效时间,所以也没有设置过期时间
- boolean hasExpire = false;
- JSONObject period = effectTime.getJSONObject("period");
- if (ObjectUtil.isAllNotEmpty(effectTime, effectTime.getString("type"))) {
- if ("period".equals(effectTime.getString("type"))) {
- //开始结束时间都不为空
- String startTime = period.getString("startTime");
- String endTime = period.getString("endTime");
- String dateTimeFormat = DateUtils.getSdfTimeNotDate(dateTime);
- if (ObjectUtil.isAllNotEmpty(startTime, endTime)) {
- if (!"000000".equals(startTime) || !"235959".equals(endTime)) {
- hasExpire = true;
- }
- if (startTime.compareTo(dateTimeFormat) > 0 || endTime.compareTo(DateUtils.getSdfTimeNotDate(dateTime)) < 0) {
- inEffectTime = false;
- }
- }
- }
- }
- if (inEffectTime) {
- //之前是报警状态,现在还是报警,报警值没变化
- //之前是是正常时候
- if (AlarmState.State.NORMAL.getType().equals(alarmState.getState())) {
- handlerAlarmWithLock(alarmDefine, alarmState, dateTime, condition, defineId, paramMap, effectTime, hasExpire, period);
- }
- } else {
- //不在生效时间的段的产生报警,要保存最近一次的数值是否是是报警
- //其他条件全部改成默认值(不报警,不过期,报警开始时间和结束时间为空)
- alarmState.reset();
- }
- alarmInfoCache.setAlarmState(defineId, alarmState);
- }
- /**
- * @param alarmDefine:
- * @param alarmState:
- * @param dateTime:
- * @param condition:
- * @param defineId:
- * @param paramMap:
- * @param effectTime:
- * @param hasExpire:
- * @param period:
- * @description: 生成报警记录和报警恢复消息(对报警定义加锁)
- * @author: LuoGuangyi
- * @createTime: 2020/12/28 18:21
- * @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.withNano(0));
- expireDate = DateUtils.localDateTime2Date(expireDateTime);
- }
- if (timeSecond >= condition.getTriggerUphold()) {
- log.warn("大于持续时间了,产生一条报警[{}]>[{}]", timeSecond, condition.getTriggerUphold());
- AlarmRecord alarmRecord = AlarmRecord.builder()
- .objType(alarmDefine.getObjType())
- .concern(alarmDefine.getConcern())
- .level(alarmDefine.getLevel())
- .projectId(CommonConst.projectId)
- .state(1)
- //报警时间为第一次满足报警条件时候的时间
- .triggerTime(DateUtils.parse(alarmState.getAlarmStartTime()))
- .treatState(INTEGER_ONE)
- .remark(alarmDefine.getRemark())
- .triggerInfo(JSONObject.toJSONString(paramMap))
- .condition(condition.toString())
- .effectEndTime(expireDate)
- .groupCode(CommonConst.groupCode)
- .itemCode(alarmDefine.getItemCode())
- .objId(alarmDefine.getObjId())
- .itemId(alarmDefine.getItemId())
- .classCode(alarmDefine.getClassCode())
- .createUser(CommonConst.systemId)
- .ibmsClassCode(alarmDefine.getIbmsClassCode())
- .ibmsSceneCode(alarmDefine.getIbmsSceneCode())
- .build();
- NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(5);
- nettyMessage.setContent(Arrays.asList(alarmRecord));
- //推送一条报警记录给远端
- nettyClient.sendMessage(nettyMessage);
- ZktAlarmRecordDO zktAlarmRecordDO = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO());
- zktAlarmRecordDO.setDefinitionId(defineId);
- zktAlarmRecordDO.setObjId(alarmDefine.getObjId());
- zktAlarmRecordDO.setItemCode(alarmDefine.getItemCode());
- zktAlarmRecordDO.setItemId(alarmDefine.getItemId());
- 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())
- .itemId(alarmDefine.getItemId())
- .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());
- }
- }
- }
|