AlarmHandleServiceImpl.java 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444
  1. package com.persagy.service.impl;
  2. import static org.apache.commons.lang3.math.NumberUtils.INTEGER_ONE;
  3. import java.time.LocalDate;
  4. import java.time.LocalDateTime;
  5. import java.time.LocalTime;
  6. import java.time.format.DateTimeFormatter;
  7. import java.util.Arrays;
  8. import java.util.Date;
  9. import java.util.HashMap;
  10. import java.util.List;
  11. import java.util.Objects;
  12. import java.util.Optional;
  13. import org.apache.commons.lang3.StringUtils;
  14. import org.quartz.JobDataMap;
  15. import org.quartz.SchedulerException;
  16. import org.springframework.beans.factory.annotation.Autowired;
  17. import org.springframework.stereotype.Service;
  18. import com.alibaba.fastjson.JSONObject;
  19. import com.googlecode.aviator.AviatorEvaluator;
  20. import com.googlecode.aviator.Expression;
  21. import com.persagy.cache.AlarmInfoCache;
  22. import com.persagy.cache.CurrentDataCache;
  23. import com.persagy.constant.CommonConst;
  24. import com.persagy.entity.AlarmDefine;
  25. import com.persagy.entity.AlarmRecord;
  26. import com.persagy.entity.AlarmState;
  27. import com.persagy.entity.Condition;
  28. import com.persagy.entity.ExpireAlarmMessage;
  29. import com.persagy.entity.NettyMessage;
  30. import com.persagy.entity.ZktAlarmRecordDO;
  31. import com.persagy.job.ExpireAlarmQueue;
  32. import com.persagy.netty.client.NettyClient;
  33. import com.persagy.repository.AlarmRecordRepository;
  34. import com.persagy.service.AlarmHandleService;
  35. import com.persagy.service.AlarmQuartzService;
  36. import com.persagy.utils.DateUtils;
  37. import com.persagy.utils.StringUtil;
  38. import cn.hutool.core.date.DateUtil;
  39. import cn.hutool.core.util.ObjectUtil;
  40. import lombok.extern.slf4j.Slf4j;
  41. /**
  42. * @description:报警处理实现类:包含报警产生和报警恢复
  43. * @author:LuoGuanhyi
  44. * @company:PersagyTechnologyCo.,Ltd
  45. * @since:2020/10/15 001511:14
  46. * @version:V1.0
  47. **/
  48. @Service
  49. @Slf4j
  50. public class AlarmHandleServiceImpl implements AlarmHandleService {
  51. @Autowired
  52. AlarmInfoCache alarmInfoCache;
  53. @Autowired
  54. CurrentDataCache currentDataCache;
  55. @Autowired
  56. NettyClient nettyClient;
  57. @Autowired
  58. AlarmQuartzService alarmQuartzService;
  59. @Autowired
  60. AlarmRecordRepository alarmRecordRepository;
  61. /**
  62. * @param msg:
  63. * @description:处理iot采集数据
  64. * @exception:
  65. * @author: LuoGuangyi
  66. * @company: Persagy Technology Co.,Ltd
  67. * @return: void
  68. * @since: 2020/10/27 14:28
  69. * @version: V1.0
  70. */
  71. @Override
  72. public void handleIOTData(String msg) throws SchedulerException, InterruptedException {
  73. JSONObject parseObject = JSONObject.parseObject(msg);
  74. msg = parseObject.getString("data");
  75. String[] split = msg.split(";");
  76. if (split.length % 4 != 0) {
  77. return;
  78. }
  79. //返回的信息点个数
  80. int nums = (split.length) / 4;
  81. for (int i = 0; i < nums; i++) {
  82. String dateTime = split[i * 4];
  83. String meterId = split[i * 4 + 1];
  84. String funcId = split[i * 4 + 2];
  85. double value = Double.parseDouble(split[i * 4 + 3]);
  86. if (alarmInfoCache.hasKey(meterId, funcId)) {
  87. validIotData(dateTime, meterId, funcId, value);
  88. }
  89. }
  90. }
  91. private void validIotData(String dateTime, String meterId, String funcId, double value) throws SchedulerException, InterruptedException {
  92. List<AlarmDefine> alarmDefines = alarmInfoCache.getAlarmDefinitionIdByMeterFuncId(meterId, funcId);
  93. for (AlarmDefine alarmDefine : alarmDefines) {
  94. String defineId = AlarmInfoCache.getAlarmDefineId(alarmDefine);
  95. synchronized (defineId.intern()) {
  96. currentDataCache.putCurrentData(meterId, funcId, value);
  97. Condition condition = alarmDefine.getCondition();
  98. List<JSONObject> codeDetail = condition.getInfoCodes();
  99. boolean match = codeDetail.stream().allMatch(p -> currentDataCache.hasKey(p.getString("meterId"), p.getString("funcId")));
  100. //报警定义的所有信息点都有采集数值,具备判断条件
  101. if (match) {
  102. if(condition.getRealRestraintTime()!=null) {
  103. if (condition.getRealRestraintTime().compareTo(LocalDateTime.now()) > 0) {
  104. continue;
  105. }
  106. }
  107. String trigger = condition.getTrigger();
  108. String end = condition.getEnd();
  109. HashMap paramMap = new HashMap<>();
  110. for (int j = 0; j < codeDetail.size(); j++) {
  111. JSONObject code = codeDetail.get(j);
  112. paramMap.put(code.getString("infoCode"), currentDataCache.getCurrentData(code.getString("meterId"), code.getString("funcId")));
  113. }
  114. Expression triggerExp = AviatorEvaluator.compile(trigger, true);
  115. Expression endExp = AviatorEvaluator.compile(end, true);
  116. Boolean triggerResult = (Boolean) triggerExp.execute(paramMap);
  117. Boolean endResult = (Boolean) endExp.execute(paramMap);
  118. log.info("triggerResult:[{}],endResult:[{}]", triggerResult, endResult);
  119. if (triggerResult && endResult) {
  120. log.warn("报警触发条件和报警恢复条件同时满足,请检查,报警定义详情【{}】", alarmDefine.toString());
  121. }
  122. AlarmState alarmState = alarmInfoCache.getAlarmState(defineId);
  123. if (Objects.isNull(alarmState)) {
  124. //默认正常报警状态
  125. alarmState = new AlarmState(defineId);
  126. Optional<ZktAlarmRecordDO> recordOptional = alarmRecordRepository.findById(defineId);
  127. if (recordOptional.isPresent()) {
  128. ZktAlarmRecordDO alarmRecordDO = recordOptional.get();
  129. //数据库报警状态:1-未处理
  130. if ("1".equals(alarmRecordDO.getState())) {
  131. alarmState.setState(AlarmState.State.NOT_DEAL.getType());
  132. alarmState.setAlarmStartTime(alarmRecordDO.getAlarmTime());
  133. }
  134. }
  135. }
  136. //报警产生值满足(这里的满足不考虑报警持续时间)
  137. if (triggerResult) {
  138. log.info("有一条满足报警条件{},-----{}----{}", defineId, paramMap, alarmDefine.getCondition());
  139. if (alarmDefine.getOpen() == 0) {
  140. log.info("报警定义ID为[{}]已经屏蔽", defineId);
  141. continue;
  142. }
  143. if (AlarmInfoCache.isolationSystemList.contains(alarmDefine.getSystemCode())) {
  144. log.info("报警定义ID为[{}]的系统[{}]已经隔离,不产生报警", defineId, alarmDefine.getSystemCode());
  145. continue;
  146. }
  147. //报警的时候不考虑报警恢复,因为同时报警和报警恢复是不应该出现的
  148. handlerNowDataAlarm(alarmDefine, alarmState, dateTime, condition, defineId, paramMap, meterId, funcId, value);
  149. } else {
  150. log.info("不满足报警条件{},{}----{}", defineId, paramMap, alarmDefine.getCondition());
  151. //当前数据正常
  152. handlerNowDataNormal(alarmDefine, dateTime, condition, defineId, endResult, alarmState, paramMap, meterId, funcId, value);
  153. }
  154. } else {
  155. log.warn("部分信息点没有数值:[{}]", codeDetail);
  156. }
  157. }
  158. }
  159. }
  160. /**
  161. * @param alarmDefine : 报警定义
  162. * @param dateTime :IOT采集时间
  163. * @param condition :报警条件
  164. * @param defineId :报警定义ID
  165. * @param endResult :报警恢复监测结果
  166. * @param alarmState : 报警状态时间实体
  167. * @param paramMap 报警恢复触发值
  168. * @param meterId: 表号
  169. * @param funcId: 功能号
  170. * @param value: 数值
  171. * @description:当前数据正常判断逻辑
  172. * @exception:
  173. * @author: LuoGuangyi
  174. * @company: Persagy Technology Co.,Ltd
  175. * @return: void
  176. * @since: 2020/10/21 16:46
  177. * @version: V1.0
  178. */
  179. 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 {
  180. //当前数据正常,报警状态为正常:清空之前的报警计时,重置回默认状态
  181. if (AlarmState.State.NORMAL.getType().equals(alarmState.getState())) {
  182. alarmState = new AlarmState(defineId);
  183. alarmInfoCache.setAlarmState(defineId, alarmState);
  184. } else if (AlarmState.State.NOT_DEAL.getType().equals(alarmState.getState())) {
  185. handlerNormalWithLock(alarmDefine, dateTime, condition, defineId, endResult, alarmState, paramMap);
  186. }
  187. }
  188. private void handlerNormalWithLock(AlarmDefine alarmDefine, String dateTime, Condition condition, String defineId, Boolean endResult, AlarmState alarmState, HashMap paramMap) throws InterruptedException {
  189. //报警状态异常时候
  190. alarmState.setLatestDataNormalstate(true);
  191. //报警恢复条件满足,判断是否满足报警恢复持续时间
  192. if (endResult) {
  193. String endTime = alarmState.getAlarmEndTime();
  194. if (StringUtil.isEmpty(endTime)) {
  195. endTime = dateTime;
  196. }
  197. alarmState.setAlarmEndTime(endTime);
  198. //设置开始恢复时间
  199. int uphold = condition.getEndUphold();
  200. //超过报警恢复设置的持续时间
  201. if (DateUtils.betweenTwoTimeSecond(endTime, dateTime) >= uphold) {
  202. log.info("产生一条报警恢复消息[{}]>[{}]", DateUtils.betweenTwoTimeSecond(endTime, dateTime), uphold);
  203. NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(6);
  204. ZktAlarmRecordDO alarmRecordDO = alarmRecordRepository.findById(AlarmInfoCache.getAlarmDefineId(alarmDefine)).orElse(new ZktAlarmRecordDO());
  205. alarmRecordDO.setDefinitionId(defineId);
  206. alarmRecordDO.setObjId(alarmDefine.getObjId());
  207. alarmRecordDO.setItemCode(alarmDefine.getItemCode());
  208. alarmRecordDO.setItemId(alarmDefine.getItemId());
  209. alarmRecordDO.setState("2");
  210. alarmRecordDO.setEndTime(DateUtils.parseDate(dateTime));
  211. alarmRecordDO.setEndInfo(JSONObject.toJSONString(paramMap));
  212. //更新报警状态
  213. alarmRecordRepository.save(alarmRecordDO);
  214. String alarmId = alarmRecordDO.getAlarmId();
  215. //报警恢复参数
  216. AlarmRecord alarmResumeRecord = AlarmRecord.builder()
  217. .state(2)
  218. .groupCode(CommonConst.groupCode)
  219. .projectId(CommonConst.projectId)
  220. .endTime(DateUtils.parse(dateTime))
  221. .endInfo(JSONObject.toJSONString(paramMap))
  222. .build();
  223. //如果有报警ID,直接报警恢复
  224. if (StringUtils.isNotEmpty(alarmId)) {
  225. alarmResumeRecord.setId(alarmId);
  226. nettyMessage.setContent(Arrays.asList(alarmResumeRecord));
  227. //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123","endTime":"","endInfo":""}
  228. nettyClient.sendMessage(nettyMessage);
  229. } else {
  230. //如果没有报警ID,定时任务再次测试
  231. JobDataMap jobDataMap = new JobDataMap();
  232. jobDataMap.put("alarmRecord", alarmRecordDO.toString());
  233. jobDataMap.put("refire", "0");
  234. jobDataMap.put("endTime", dateTime);
  235. jobDataMap.put("endInfo", JSONObject.toJSONString(paramMap));
  236. jobDataMap.put("defineId", defineId);
  237. //恢复
  238. jobDataMap.put("state", "2");
  239. log.info(JSONObject.toJSONString(jobDataMap));
  240. String jobName = defineId;
  241. ExpireAlarmMessage em = new ExpireAlarmMessage();
  242. //过期消息
  243. em.setType("1");
  244. em.setStartTime(DateUtil.offsetMinute(new Date(), 3).toJdkDate());
  245. em.setJobDataMap(jobDataMap);
  246. em.setJobName(jobName);
  247. em.setJobGroupName("resume");
  248. ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
  249. }
  250. String jobName = AlarmInfoCache.getAlarmDefineId(alarmDefine);
  251. ExpireAlarmMessage em = new ExpireAlarmMessage();
  252. //取消过期消息
  253. em.setType("2");
  254. em.setJobName(jobName);
  255. em.setJobGroupName("expire");
  256. ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
  257. //报警恢复,报警状态重置回默认
  258. alarmState.reset();
  259. }
  260. }
  261. alarmInfoCache.setAlarmState(defineId, alarmState);
  262. }
  263. /**
  264. * @param alarmDefine 报警定义
  265. * @param dateTime :IOT数据采集时间
  266. * @param condition :报警定义条件
  267. * @param defineId :报警定义ID
  268. * @param paramMap 报警触发值
  269. * @param meterId 表号
  270. * @param funcId 功能号
  271. * @param value IOT采集数值
  272. * @description:处理当前值报警的情况
  273. * @exception:
  274. * @author: LuoGuangyi
  275. * @company: Persagy Technology Co.,Ltd
  276. * @return: void
  277. * @since: 2020/10/20 22:31
  278. * @version: V1.0
  279. */
  280. private void handlerNowDataAlarm(AlarmDefine alarmDefine, AlarmState alarmState, String dateTime, Condition condition, String defineId, HashMap paramMap, String meterId, String funcId, double value) throws SchedulerException, InterruptedException {
  281. alarmState.setLatestDataNormalstate(false);
  282. JSONObject effectTime = condition.getEffectTime();
  283. //无生效时间设置,则任何时间生效
  284. boolean inEffectTime = true;
  285. //没有设置生效时间,所以也没有设置过期时间
  286. boolean hasExpire = false;
  287. JSONObject period = effectTime.getJSONObject("period");
  288. if (ObjectUtil.isAllNotEmpty(effectTime, effectTime.getString("type"))) {
  289. if ("period".equals(effectTime.getString("type"))) {
  290. //开始结束时间都不为空
  291. String startTime = period.getString("startTime");
  292. String endTime = period.getString("endTime");
  293. String dateTimeFormat = DateUtils.getSdfTimeNotDate(dateTime);
  294. if (ObjectUtil.isAllNotEmpty(startTime, endTime)) {
  295. if (!"000000".equals(startTime) || !"235959".equals(endTime)) {
  296. hasExpire = true;
  297. }
  298. if (startTime.compareTo(dateTimeFormat) > 0 || endTime.compareTo(DateUtils.getSdfTimeNotDate(dateTime)) < 0) {
  299. inEffectTime = false;
  300. }
  301. }
  302. }
  303. }
  304. if (inEffectTime) {
  305. //之前是报警状态,现在还是报警,报警值没变化
  306. //之前是是正常时候
  307. if (AlarmState.State.NORMAL.getType().equals(alarmState.getState())) {
  308. handlerAlarmWithLock(alarmDefine, alarmState, dateTime, condition, defineId, paramMap, effectTime, hasExpire, period);
  309. }
  310. } else {
  311. //不在生效时间的段的产生报警,要保存最近一次的数值是否是是报警
  312. //其他条件全部改成默认值(不报警,不过期,报警开始时间和结束时间为空)
  313. alarmState.reset();
  314. }
  315. alarmInfoCache.setAlarmState(defineId, alarmState);
  316. }
  317. /**
  318. * @param alarmDefine:
  319. * @param alarmState:
  320. * @param dateTime:
  321. * @param condition:
  322. * @param defineId:
  323. * @param paramMap:
  324. * @param effectTime:
  325. * @param hasExpire:
  326. * @param period:
  327. * @description: 生成报警记录和报警恢复消息(对报警定义加锁)
  328. * @author: LuoGuangyi
  329. * @createTime: 2020/12/28 18:21
  330. * @return: void
  331. * @expression
  332. */
  333. private void handlerAlarmWithLock(AlarmDefine alarmDefine, AlarmState alarmState, String dateTime, Condition condition, String defineId, HashMap paramMap, JSONObject effectTime, boolean hasExpire, JSONObject period) throws InterruptedException {
  334. long timeSecond = 0;
  335. if (StringUtil.isNotEmpty(alarmState.getAlarmStartTime())) {
  336. timeSecond = DateUtils.betweenTwoTimeSecond(alarmState.getAlarmStartTime(), dateTime);
  337. } else {
  338. //设置开始报警时间
  339. alarmState.setAlarmStartTime(dateTime);
  340. }
  341. LocalDateTime expireDateTime = null;
  342. Date expireDate = null;
  343. if (hasExpire && "period".equals(effectTime.getString("type"))) {
  344. //过期时间
  345. String expireTime = period.getString("endTime");
  346. LocalTime localTime = LocalTime.parse(expireTime, DateTimeFormatter.ofPattern(DateUtils.sdfTimeNotDate));
  347. expireDateTime = LocalDateTime.of(LocalDate.now(), localTime.withNano(0));
  348. expireDate = DateUtils.localDateTime2Date(expireDateTime);
  349. }
  350. if (timeSecond >= condition.getTriggerUphold()) {
  351. log.warn("大于持续时间了,产生一条报警[{}]>[{}]", timeSecond, condition.getTriggerUphold());
  352. AlarmRecord alarmRecord = AlarmRecord.builder()
  353. .objType(alarmDefine.getObjType())
  354. .concern(alarmDefine.getConcern())
  355. .level(alarmDefine.getLevel())
  356. .projectId(CommonConst.projectId)
  357. .state(1)
  358. //报警时间为第一次满足报警条件时候的时间
  359. .triggerTime(DateUtils.parse(alarmState.getAlarmStartTime()))
  360. .treatState(INTEGER_ONE)
  361. .remark(alarmDefine.getRemark())
  362. .triggerInfo(JSONObject.toJSONString(paramMap))
  363. .condition(condition.toString())
  364. .effectEndTime(expireDate)
  365. .groupCode(CommonConst.groupCode)
  366. .itemCode(alarmDefine.getItemCode())
  367. .objId(alarmDefine.getObjId())
  368. .itemId(alarmDefine.getItemId())
  369. .classCode(alarmDefine.getClassCode())
  370. .createUser(CommonConst.systemId)
  371. .ibmsClassCode(alarmDefine.getIbmsClassCode())
  372. .ibmsSceneCode(alarmDefine.getIbmsSceneCode())
  373. .build();
  374. NettyMessage<AlarmRecord> nettyMessage = new NettyMessage<>(5);
  375. nettyMessage.setContent(Arrays.asList(alarmRecord));
  376. //推送一条报警记录给远端
  377. nettyClient.sendMessage(nettyMessage);
  378. ZktAlarmRecordDO zktAlarmRecordDO = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO());
  379. zktAlarmRecordDO.setDefinitionId(defineId);
  380. zktAlarmRecordDO.setObjId(alarmDefine.getObjId());
  381. zktAlarmRecordDO.setItemCode(alarmDefine.getItemCode());
  382. zktAlarmRecordDO.setItemId(alarmDefine.getItemId());
  383. zktAlarmRecordDO.setState("1");
  384. zktAlarmRecordDO.setAlarmTime(alarmState.getAlarmStartTime());
  385. zktAlarmRecordDO.setProjectId(alarmDefine.getProjectId());
  386. alarmRecordRepository.save(zktAlarmRecordDO);
  387. alarmState.setState(AlarmState.State.NOT_DEAL.getType());
  388. //有过期时间,生成报警过期消息
  389. if (hasExpire && "period".equals(effectTime.getString("type"))) {
  390. //过期时间
  391. log.error("产生一条定时过期报警消息");
  392. String alarmId = alarmRecordRepository.findById(defineId).orElse(new ZktAlarmRecordDO()).getAlarmId();
  393. ZktAlarmRecordDO alarmRecordDO = ZktAlarmRecordDO.builder()
  394. .alarmTime(alarmState.getAlarmStartTime())
  395. .effectEndTime(DateUtils.format(expireDateTime))
  396. .definitionId(defineId)
  397. .itemCode(alarmDefine.getItemCode())
  398. .name(alarmDefine.getName())
  399. .objId(alarmDefine.getObjId())
  400. .itemId(alarmDefine.getItemId())
  401. .projectId(alarmDefine.getProjectId())
  402. .state("3") //要变成已过期
  403. .alarmId(alarmId)
  404. .build();
  405. JobDataMap jobDataMap = new JobDataMap();
  406. jobDataMap.put("alarmRecord", alarmRecordDO.toString());
  407. jobDataMap.put("refire", "0");
  408. jobDataMap.put("expireTime", DateUtils.format(expireDateTime));
  409. jobDataMap.put("defineId", defineId);
  410. //过期
  411. jobDataMap.put("state", "3");
  412. String jobName = defineId;
  413. ExpireAlarmMessage em = new ExpireAlarmMessage();
  414. //过期消息
  415. em.setType("1");
  416. em.setStartTime(DateUtils.localDateTime2Date(expireDateTime));
  417. em.setJobDataMap(jobDataMap);
  418. em.setJobName(jobName);
  419. em.setJobGroupName("expire");
  420. ExpireAlarmQueue.getExpireAlarmMessageQueue().produce(em);
  421. //alarmQuartzService.addExpireJob(DateUtils.localDateTime2Date(expireDateTime), jobName, "expire", jobDataMap);
  422. }
  423. } else {
  424. alarmState.setState(AlarmState.State.NORMAL.getType());
  425. }
  426. }
  427. }