Browse Source

修改mq发送报警消息

xujiaheng 3 years ago
parent
commit
cdb4ae6913

+ 11 - 3
src/main/java/com/persagy/job/AlarmExpireJob.java

@@ -5,6 +5,7 @@ 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.cache.AlarmRedisCache;
 import com.persagy.cache.CreatedAlarmIdsCache;
 import com.persagy.client.GroupNettyClient;
 import com.persagy.entity.AlarmConditionState;
@@ -15,6 +16,8 @@ import com.persagy.entity.v2.DmpMessage;
 import com.persagy.enumeration.AlarmDefineStateEnum;
 import com.persagy.enumeration.NettyMsgTypeEnum;
 import com.persagy.repository.AlarmRecordRepository;
+import com.persagy.service.AlarmHandleService;
+import com.persagy.service.impl.AlarmHandleServiceImpl;
 import com.persagy.utils.DateUtils;
 import com.persagy.utils.StringUtil;
 import lombok.extern.slf4j.Slf4j;
@@ -44,6 +47,10 @@ public class AlarmExpireJob extends QuartzJobBean {
     AlarmInfoCache alarmInfoCache;
     @Autowired
     CreatedAlarmIdsCache createdAlarmIdsCache;
+    @Autowired
+    AlarmHandleService alarmHandleService;
+    @Autowired
+    AlarmRedisCache alarmRedisCache;
     /**
      * 报警记录信息详情
      */
@@ -165,9 +172,9 @@ public class AlarmExpireJob extends QuartzJobBean {
                 //{"id","123", "state":1, "groupCode":"wd", "projectId":"Pj123"}
                 log.info("定时任务执行完成,向云端推送{}消息", jobType);
                 // TODO: 2021/11/17 向mq推送消息
-                // TODO: 2021/11/17 移除缓存信息,将报警定义状态更新到redis
+                alarmHandleService.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(dmpMessage)),alarmId,zktAlarmRecordDO.getProjectId());
                 // 删除缓存中报警id
-                createdAlarmIdsCache.remove(alarmId);
+                //createdAlarmIdsCache.remove(alarmId);
 
                 String defineId = zktAlarmRecordDO.getDefinitionId();
                 synchronized (defineId.intern()) {
@@ -187,7 +194,8 @@ public class AlarmExpireJob extends QuartzJobBean {
                     if (AlarmDefineStateEnum.EXPIRED.getValue().equals(state)) {
                         AlarmConditionState alarmConditionState = new AlarmConditionState(this.defineId);
                         alarmConditionState.setAlarmStartTime("");
-                        alarmInfoCache.setAlarmState(this.defineId, alarmConditionState);
+                        // TODO: 2021/11/17 移除缓存信息,将报警定义状态更新到redis
+                        alarmRedisCache.setAlarmState(this.defineId, alarmConditionState);
                     }
 
                     // 如果报警仍是创建任务时的报警,数据库中删除这条报警定义

+ 2 - 2
src/main/java/com/persagy/mq/RabbitConfig.java

@@ -23,12 +23,12 @@ public class RabbitConfig {
 	/**
 	 * 报警消息路由键
 	 */
-	public static final String alarmMsgRoutingKey = "alarm-msg-routing-key";
+	public static final String alarmMsgRoutingKey = "alarm-record-routing-key";
 
 	/**
 	 * 报警条件队列
 	 */
-	private final String alarmMsgQueue = "alarm-msg-queue";
+	private final String alarmMsgQueue = "alarm-record-queue";
 
 	@Bean
 	public Queue alarmMsgQueue() {

+ 3 - 0
src/main/java/com/persagy/service/AlarmHandleService.java

@@ -1,5 +1,6 @@
 package com.persagy.service;
 
+import com.alibaba.fastjson.JSONObject;
 import org.quartz.SchedulerException;
 
 /**
@@ -22,4 +23,6 @@ public interface AlarmHandleService {
      * @version V1.0
      */
     void handleIotMsg(String msg) throws Exception;
+
+    void sendToMQ(JSONObject parseObject, String alarmId, String projectId);
 }

+ 53 - 37
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -609,13 +609,14 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 //                    alarmLastTimeCache.put(alarmId, Integer.valueOf(String.valueOf(lastTime)));
 //                    log.debug("开始拼装报警持续消息");
 //                    initAlarmContinueMsg(alarmId, lastTime, paramMap, alarmDefine.getProjectId());
-					ObjectMapper objectMapper = new ObjectMapper();
-					ObjectNode objectNode = objectMapper.createObjectNode();
-					objectNode.put("id", alarmId);
-					objectNode.put("lastTime", lastTime);
-					objectNode.set("iotData", objectMapper.readTree(objectMapper.writeValueAsString(paramMap)));
-					objectNode.put("groupCode", groupCode);
-					this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(objectNode)), alarmId, alarmDefine);
+                    ObjectMapper objectMapper = new ObjectMapper();
+                    ObjectNode objectNode = objectMapper.createObjectNode();
+                    objectNode.put("id", alarmId);
+                    objectNode.put("lastTime", lastTime);
+                    objectNode.set("iotData", objectMapper.readTree(objectMapper.writeValueAsString(paramMap)));
+                    objectNode.put("groupCode", groupCode);
+					//推送报警持续消息到MQ
+					this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(objectNode)),alarmId,alarmDefine.getProjectId());
 				}
 			}
 			// 之前是是正常状态
@@ -662,7 +663,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 			log.debug("生效时间为:[{}]", condition.getEffectTime());
 			alarmConditionState.reset();
 		}
-		alarmInfoCache.setAlarmState(defineId, alarmConditionState);
+		alarmRedisCache.setAlarmState(defineId, alarmConditionState);
 	}
 
 	/**
@@ -774,39 +775,54 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 		return zktAlarmRecordDO;
 	}
 
-	/**
-	 * @description: 发送创建报警记录消息
-	 * @param: alarmId     报警id
-	 * @param: alarmDefine 报警定义
-	 * @param: triggerTime 触发时间
-	 * @param: triggerInfo 触发值
-	 * @param: expireDate  过期时间
-	 * @return: void
-	 * @exception:
-	 * @author: lixing
-	 * @company: Persagy Technology Co.,Ltd
-	 * @since: 2021/1/8 下午5:38
-	 * @version: V1.0
-	 */
-	private void sendCreateAlarmRecordMessage(String alarmId, AlarmDefine alarmDefine, String triggerTime,
-			HashMap<String, Object> triggerInfo, Date expireDate) throws InterruptedException {
-		Condition condition = alarmDefine.getCondition();
-		AlarmRecord alarmRecord = AlarmRecord.builder().id(alarmId).category(alarmDefine.getCategory())
-				.concern(alarmDefine.getConcern()).level(alarmDefine.getLevel()).projectId(alarmDefine.getProjectId())
-				.state(1).triggerTime(DateUtils.parseDate(triggerTime)).remark(alarmDefine.getRemark())
-				.triggerInfo(JSONObject.toJSONString(triggerInfo)).condition(condition.toString())
-				.effectEndTime(expireDate).groupCode(groupCode).itemCode(alarmDefine.getItemCode())
-				.objId(alarmDefine.getObjId()).classCode(alarmDefine.getClassCode()).createUser(systemId).treatState(1)
-				.build();
-		// TODO: 2021/11/16 消息发送至mq
-		this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(alarmRecord)), alarmId, alarmDefine);
-	}
 
-	public void sendToMQ(JSONObject message, String alarmId, AlarmDefine alarmDefine) {
+    /**
+     * @description: 发送创建报警记录消息
+     * @param: alarmId 报警id
+     * @param: alarmDefine 报警定义
+     * @param: triggerTime 触发时间
+     * @param: triggerInfo 触发值
+     * @param: expireDate 过期时间
+     * @return: void
+     * @exception:
+     * @author: lixing
+     * @company: Persagy Technology Co.,Ltd
+     * @since: 2021/1/8 下午5:38
+     * @version: V1.0
+     */
+    private void sendCreateAlarmRecordMessage(
+            String alarmId,
+            AlarmDefine alarmDefine, String triggerTime,
+            HashMap<String, Object> triggerInfo, Date expireDate) throws InterruptedException {
+        Condition condition = alarmDefine.getCondition();
+        AlarmRecord alarmRecord = AlarmRecord.builder()
+                .id(alarmId)
+                .category(alarmDefine.getCategory())
+                .concern(alarmDefine.getConcern())
+                .level(alarmDefine.getLevel())
+                .projectId(alarmDefine.getProjectId())
+                .state(1)
+                .triggerTime(DateUtils.parseDate(triggerTime))
+                .remark(alarmDefine.getRemark())
+                .triggerInfo(JSONObject.toJSONString(triggerInfo))
+                .condition(condition.toString())
+                .effectEndTime(expireDate)
+                .groupCode(groupCode)
+                .itemCode(alarmDefine.getItemCode())
+                .objId(alarmDefine.getObjId())
+                .classCode(alarmDefine.getClassCode())
+                .createUser(systemId)
+                .treatState(1)
+                .build();
+        // TODO: 2021/11/16 消息发送至mq
+		this.sendToMQ(JSONObject.parseObject(JSONObject.toJSONString(alarmRecord)),alarmId,alarmDefine.getProjectId());
+    }
+
+	public void sendToMQ(JSONObject message,String alarmId,String projectId){
 		DmpMessage dmpMessage = new DmpMessage();
 		dmpMessage.setDefaultValue();
 		dmpMessage.setMid(alarmId);
-		dmpMessage.setProjectId(alarmDefine.getProjectId());
+		dmpMessage.setProjectId(projectId);
 		dmpMessage.setGroupCode(groupCode);
 		dmpMessage.setExts(message);
 		rabbitTemplate.convertAndSend(RabbitConfig.exchange, RabbitConfig.alarmMsgRoutingKey,