gaoyu 3 rokov pred
rodič
commit
01eeea3001

+ 1 - 3
src/main/java/com/persagy/ZktProjectAlarmApplication.java

@@ -1,12 +1,10 @@
 package com.persagy;
 
-import com.github.dadiyang.autologging.aop.annotation.AutoLogAll;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
-import java.util.Scanner;
+import com.github.dadiyang.autologging.aop.annotation.AutoLogAll;
 
 @SpringBootApplication
 @EnableScheduling

+ 28 - 15
src/main/java/com/persagy/cache/AlarmRedisCache.java

@@ -1,6 +1,7 @@
 package com.persagy.cache;
 
-import java.util.Objects;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
@@ -13,14 +14,17 @@ import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.persagy.constant.RedisConstant;
 import com.persagy.entity.AlarmConditionState;
 import com.persagy.entity.ZktAlarmRecordDO;
 import com.persagy.entity.v2.AlarmCondition;
+import com.persagy.entity.v2.ItemCodeCondition;
 import com.persagy.entity.v2.ObjConditionInfo;
 import com.persagy.repository.AlarmRecordRepository;
 
+import cn.hutool.json.JSONUtil;
 import lombok.extern.slf4j.Slf4j;
 
 @Component
@@ -35,17 +39,15 @@ public class AlarmRedisCache implements RedisCache {
 
 	public AlarmConditionState getAlarmDefineState(String defineId) {
 		AlarmConditionState alarmConditionState = null;
-		String str = alarmRedisTemplate.opsForHash().get(RedisConstant.DEFINE_ID, defineId).toString();
-		if (StringUtils.isNotBlank(str)) {
-			try {
-				JSONObject object = JSONObject.parseObject(str);
-				alarmConditionState = JSON.toJavaObject(object, AlarmConditionState.class);
-			} catch (Exception e) {
-				log.error(" alarmDefineState json 转换 AlarmConditionState 失败", e);
-			}
-		}
 		try {
-			if (Objects.isNull(alarmConditionState)) {
+			Object obj = alarmRedisTemplate.opsForHash().get(RedisConstant.DEFINE_ID, defineId);
+			if (null != obj && StringUtils.isNotBlank(obj.toString())) {
+				try {
+					alarmConditionState =JSONUtil.toBean(obj.toString(), AlarmConditionState.class);
+				} catch (Exception e) {
+					log.error(" alarmDefineState json 转换 AlarmConditionState 失败", e);
+				}
+			} else {
 				// 默认正常报警状态
 				Optional<ZktAlarmRecordDO> recordOptional = alarmRecordRepository.findById(defineId);
 				if (recordOptional.isPresent()) {
@@ -56,6 +58,8 @@ public class AlarmRedisCache implements RedisCache {
 						alarmConditionState.setState(AlarmConditionState.State.NOT_DEAL.getType());
 						alarmConditionState.setAlarmStartTime(alarmRecordDO.getAlarmTime());
 					}
+				} else {
+					alarmConditionState = new AlarmConditionState(defineId);
 				}
 			}
 			log.debug("alarmDefineState: [{}], ( 0-正常 1-报警)", alarmConditionState.getState());
@@ -85,7 +89,13 @@ public class AlarmRedisCache implements RedisCache {
 			String str = alarmRedisTemplate.opsForHash().get(RedisConstant.METER_ID, meterId).toString();
 			if (StringUtils.isNotBlank(str)) {
 				JSONObject object = JSONObject.parseObject(str);
+				JSONArray conditions = object.getJSONArray("conditions");
+				if (null == conditions || conditions.isEmpty()) {
+					return null;
+				}
 				ObjConditionInfo objConditionInfo = JSON.toJavaObject(object, ObjConditionInfo.class);
+				List<ItemCodeCondition> iccArrayy = JSON.parseArray(conditions.toString(), ItemCodeCondition.class);
+				objConditionInfo.setConditions(new LinkedList<>(iccArrayy));
 				return objConditionInfo;
 			}
 		} catch (Exception e) {
@@ -96,9 +106,9 @@ public class AlarmRedisCache implements RedisCache {
 
 	public AlarmCondition getAlarmCondition(String conditionId) {
 		try {
-			String str = alarmRedisTemplate.opsForHash().get(RedisConstant.ALARMCONDITION_ID, conditionId).toString();
-			if (StringUtils.isNotBlank(str)) {
-				JSONObject object = JSONObject.parseObject(str);
+			Object obj = alarmRedisTemplate.opsForHash().get(RedisConstant.ALARMCONDITION_ID, conditionId);
+			if (null != obj && StringUtils.isNotBlank(obj.toString())) {
+				JSONObject object = JSONObject.parseObject(obj.toString());
 				AlarmCondition alarmCondition = JSON.toJavaObject(object, AlarmCondition.class);
 				return alarmCondition;
 			}
@@ -126,8 +136,10 @@ public class AlarmRedisCache implements RedisCache {
 	 */
 	public boolean lock(String key, String value, Integer timeout) {
 		boolean bool;
+		int i = 1;
 		while (true) {
 			Boolean b = alarmRedisTemplate.opsForValue().setIfAbsent(key, value, timeout, TimeUnit.SECONDS);
+			System.out.println(key + "---key----" + b);
 			if (b) {
 				bool = b;
 				break;
@@ -138,9 +150,10 @@ public class AlarmRedisCache implements RedisCache {
 					Thread.currentThread().interrupt();
 				}
 			}
+			System.out.println(i);
 		}
 		return bool;
 
 	}
-	
+
 }

+ 1 - 1
src/main/java/com/persagy/constant/RedisConstant.java

@@ -71,7 +71,7 @@ public class RedisConstant {
     /**
      * 报警条件Id REDIS hash 的 id
      */
-    public static final String ALARMCONDITION_ID = "ALARMCONDITION_ID";
+    public static final String ALARMCONDITION_ID = "ALARM_CONDITION_ID";
 
 
     /**

+ 1 - 1
src/main/java/com/persagy/entity/AlarmConditionState.java

@@ -19,7 +19,7 @@ public class AlarmConditionState {
     /**
      * 报警状态( 0-正常 1-报警)
      */
-    private String state = "0";
+    private String state= "0";
     /**
      * 是否过期(0-未过期,1-已过期)
      */

+ 6 - 10
src/main/java/com/persagy/entity/AlarmRecordIdsCache.java

@@ -1,22 +1,18 @@
 package com.persagy.entity;
 
 
-import com.alibaba.fastjson.JSONObject;
-import com.alibaba.fastjson.annotation.JSONField;
+import java.io.Serializable;
+
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
 import io.swagger.annotations.ApiModel;
-import io.swagger.annotations.ApiModelProperty;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Table;
-import java.io.Serializable;
-import java.util.Date;
-
 /**
  * 缓存云端返回的报警记录id
  */

+ 6 - 10
src/main/java/com/persagy/init/InitRunner.java

@@ -1,26 +1,24 @@
 package com.persagy.init;
 
 
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Service;
+
 import com.googlecode.aviator.AviatorEvaluator;
 import com.persagy.cache.AlarmLastTimeCache;
 import com.persagy.cache.CreatedAlarmIdsCache;
 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;
-import org.springframework.core.annotation.Order;
-import org.springframework.stereotype.Service;
 
 
 @Service
 @Order(1)
 @Slf4j
 public class InitRunner implements CommandLineRunner {
-
-	@Autowired
-	private WebSocketClientFactory webSocketClientFactory;
 	@Autowired
 	private GroupNettyClient groupNettyClient;
 	@Autowired
@@ -41,8 +39,6 @@ public class InitRunner implements CommandLineRunner {
 		AviatorEvaluator.getInstance().useLRUExpressionCache(10000);
 		//启动netty客户端,接受云端数据
 		groupNettyClient.start();
-		//启动websocket,接受IOT采集数据
-		webSocketClientFactory.retryOutCallWebSocketClient();
 		alarmQuartzService.initAlarmMessagehandler();
 	}
 }

+ 7 - 22
src/main/java/com/persagy/mq/RabbitConfig.java

@@ -19,30 +19,15 @@ public class RabbitConfig {
 	 * 交换机
 	 */
 	public static final String exchange = "exchange-msg-service";
-
-	/**
-	 * 报警消息路由键
-	 */
-	public static final String alarmMsgRoutingKey = "alarm-record-routing-key";
-
-	/**
-	 * 报警条件队列
-	 */
-	private final String alarmMsgQueue = "alarm-record-queue";
-
-	@Bean
-	public Queue alarmMsgQueue() {
-		return new Queue(alarmMsgQueue, true);
-	}
-
-	@Bean
-	public TopicExchange alarmServiceExchange() {
-		return new TopicExchange(exchange);
-	}
+	public static final String exchangeAlarm = "exchange-alarm-service";
+	public static final String alarmMsgRouting = "alarm-data-routing-key";
+	public static final String alarmMsgRoutingKey = "alarm-msg-routing-key";
+	private final String alarmDataQueue = "alarm-data-queue";
 
 	@Bean
-	public Binding conditionBinding() {
-		return BindingBuilder.bind(alarmMsgQueue()).to(alarmServiceExchange()).with(alarmMsgRoutingKey);
+	public Binding conditionBindingAlarm() {
+		return BindingBuilder.bind(new Queue(alarmDataQueue, true)).to(new TopicExchange(exchange))
+				.with(alarmMsgRouting);
 	}
 
 }

+ 15 - 18
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -173,7 +173,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 							}
 							triggerOrEndAlarm(dataTime, meterId, itemCode, paramMap);
 						} catch (Exception e) {
-							log.error("线程池处理发数信息发生错误",e); 
+							log.error("线程池处理发数信息发生错误", e);
 						}
 					}
 					alarmRedisCache.releaseLock(entry.getKey());
@@ -181,7 +181,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 			}
 			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
 		} catch (Exception e) {
-			log.error("mq 接收处理发数发生错误",e); 
+			log.error("mq 接收处理发数数据发生错误", e);
 		}
 
 	}
@@ -256,19 +256,15 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 				log.error("报警条件转换为报警定义出现异常", e);
 				return;
 			}
-
-			synchronized (alarmDefine.getId().intern()) {
-
-				// 报警产生值满足(这里的满足不考虑报警持续时间)
-				if (triggerResult) {
-					log.debug("满足报警条件");
-					// 报警的时候不考虑报警恢复,因为同时报警和报警恢复是不应该出现的
-					handleAlarmTriggerData(alarmDefine, dataTime, paramMap);
-				} else {
-					log.debug("不满足报警条件");
-					// 当前数据正常
-					handlerNormalData(alarmDefine, dataTime, endResult, paramMap);
-				}
+			// 报警产生值满足(这里的满足不考虑报警持续时间)
+			if (triggerResult) {
+				log.debug("满足报警条件");
+				// 报警的时候不考虑报警恢复,因为同时报警和报警恢复是不应该出现的
+				handleAlarmTriggerData(alarmDefine, dataTime, paramMap);
+			} else {
+				log.debug("不满足报警条件");
+				// 当前数据正常
+				handlerNormalData(alarmDefine, dataTime, endResult, paramMap);
 			}
 		}
 
@@ -417,7 +413,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 			log.debug("报警恢复持续时间大于设定时间:[{}]>[{}]", alarmSuspendLastTime, uphold);
 			ZktAlarmRecordDO alarmRecordDO = getZktAlarmRecordDOWhenAlarmSuspend(defineId, dataTime, paramMap);
 			String alarmId = alarmRecordDO.getAlarmId();
-			alarmRedisCache.lock(alarmId, alarmId, 10);
+			alarmRedisCache.lock(alarmId, alarmId, 1000);
 			// 发送报警恢复消息
 			sendAlarmRecoverMessage(defineId, alarmSuspendStartTime, paramMap, alarmId);
 			// 报警恢复后,从缓存中移除报警id
@@ -489,7 +485,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 		dmpMessage.setProjectId(projectId);
 		dmpMessage.setGroupCode(groupCode);
 		dmpMessage.setExts(JSONObject.parseObject(JSONObject.toJSONString(alarmRecoverRecord)));
-		rabbitTemplate.convertAndSend(RabbitConfig.exchange, RabbitConfig.alarmMsgRoutingKey,
+		rabbitTemplate.convertAndSend(RabbitConfig.exchangeAlarm, RabbitConfig.alarmMsgRoutingKey,
 				JSON.toJSONString(dmpMessage));
 	}
 
@@ -812,7 +808,8 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
 		dmpMessage.setProjectId(projectId);
 		dmpMessage.setGroupCode(groupCode);
 		dmpMessage.setExts(message);
-		rabbitTemplate.convertAndSend(RabbitConfig.exchange, RabbitConfig.alarmMsgRoutingKey,
+		alarmRedisCache.lock(alarmId, alarmId, 1000);
+		rabbitTemplate.convertAndSend(RabbitConfig.exchangeAlarm, RabbitConfig.alarmMsgRoutingKey,
 				JSON.toJSONString(dmpMessage));
 	}
 }