添加物联告警缓存

amos-boot-data/amos-boot-data-alarm/src/main/java/com/yeejoin/amos/api/alarm/service/impl/AlarmKafkaConsumer.java

@@ -2,6 +2,7 @@ package com.yeejoin.amos.api.alarm.service.impl;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.annotation.TopicPartition;
 import org.springframework.kafka.support.Acknowledgment;
@@ -18,6 +19,9 @@ public class AlarmKafkaConsumer {
 
 	@Autowired
 	PointSystemServiceImpl pointSystemServiceImpl;
+	
+	@Value("${warning.wait.time:1000}")
+	private long waitTime;
 
 	// 消费者来处理消息
 	@KafkaListener(id = "alarmInfo", topics = { "${kafka.equipment.alarm}" })
@@ -29,7 +33,7 @@ public class AlarmKafkaConsumer {
 		pointSystemServiceImpl.sendWarningAsync(date);
 		
 		// 等待1s
-		Thread.sleep(1000L);
+		Thread.sleep(waitTime);
 		
 		// 手动提交
 		ack.acknowledge();

amos-boot-data/amos-boot-data-alarm/src/main/java/com/yeejoin/amos/api/alarm/service/impl/PointSystemServiceImpl.java

@@ -6,7 +6,6 @@ import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collector;
 import java.util.stream.Collectors;
 
 import javax.annotation.PostConstruct;
@@ -39,6 +38,7 @@ import com.yeejoin.amos.api.alarm.mapper2.JumpConfigMapper;
 import com.yeejoin.amos.api.alarm.mapper2.KKSDataMapper;
 import com.yeejoin.amos.api.alarm.mapper2.StationBasicMapper;
 import com.yeejoin.amos.api.alarm.service.IPointSystemService;
+import com.yeejoin.amos.api.alarm.utils.RedisUtils;
 
 /**
  * @description:
@@ -74,6 +74,18 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 	@Autowired
 	private StationBasicMapper stationBasicMapper;
 
+	@Autowired
+	private RedisUtils redisUtil;
+
+	@Value("${warning.redis.use:false}")
+	private boolean redisUse;
+
+	@Value("${warning.redis.limit.time:3600}")
+	private long limitTime;
+
+	@Value("${warning.SYZ:false}")
+	private boolean warnSYZ;
+
 	public String getJumpUrlByInfo(String sbbm) {
 		List<JumpConfig> jumpConfigs = jumpConfigMapper.selectList(null);
 		Map<String, String> collect = jumpConfigs.stream()
@@ -115,9 +127,9 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 
 //	@PostConstruct
 //	public void test() {
-//		String address = "16469";
+//		String address = "22163";
 //		String value = "1.0";
-//		String valueLabe = "1.0";
+//		String valueLabe = "扇区1限功率";
 //		String gatewayId = "1668801435891929089";
 //		String isAlarm = "1";
 //		this.sendWarning(address, value, valueLabe, gatewayId, isAlarm);
@@ -146,7 +158,10 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 		pointSystemWrapper.lambda().eq(PointSystem::getGatewayId, gatewayId);
 		List<PointSystem> pointSystems = pointSystemMapper.selectList(pointSystemWrapper);
 		if (pointSystems == null || pointSystems.size() < 1) {
-			throw new RuntimeException("获取kks码失败！");
+			// throw new RuntimeException("获取kks码失败: "+"address: " + address + ",gatewayId:
+			// " + gatewayId + " ,value:" + value);
+			System.out.println("获取kks码失败: " + "address:" + address + ", gatewayId:" + gatewayId + " ,value:" + value);
+			return;
 		}
 
 		PointSystem pointSystem = pointSystems.get(0);
@@ -159,10 +174,12 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 		KKSData KKSData = kksDataMapper.selectOne(KKSDataWrapper);
 		JSONObject eqdata = new JSONObject();
 		if (KKSData == null) {
-			throw new RuntimeException("kks码查询热工院表不存在:" + pointSystem.getKks());
+			// throw new RuntimeException("kks码查询热工院表不存在:" + pointSystem.getKks());
+			System.out.println("kks码查询热工院表不存在:" + pointSystem.getKks());
+			return;
 		}
 		eqdata.put("kksms", KKSData.getKKSMS());
-		
+
 		QueryWrapper<StationBasic> stationWrapper = new QueryWrapper<>();
 		stationWrapper.lambda().eq(StationBasic::getStationNumber, pointSystem.getStation());
 		StationBasic stationBasic = stationBasicMapper.selectOne(stationWrapper);
@@ -170,7 +187,9 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 			eqdata.put("sourceAttribution", stationBasic.getProjectOrgCode());
 			eqdata.put("sourceAttributionDesc", stationBasic.getStationName());
 		} else {
-			throw new RuntimeException("获取场站失败！");
+			// throw new RuntimeException("获取场站失败: " + pointSystem.getStation());
+			System.out.println("获取场站失败: " + pointSystem.getStation());
+			return;
 		}
 		try {
 //			Map<String, String> maps = new HashMap<>();
@@ -192,9 +211,11 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 //			eqdata = (JSONObject) list.get(0);
 			// 组装数据，发送预警
 			WarningDto warningDto = setWarningDto(pointSystem, eqdata, valueLabe);
-			emqKeeper.getMqttClient().publish(STATIONWARNING, JSON.toJSONString(warningDto).getBytes(), 0, false);
-			System.out.println("发送预警成功: " + JSON.toJSONString(warningDto));
-			logger.info("发送预警成功: " + JSON.toJSONString(warningDto));
+			if (warningDto != null) {
+				emqKeeper.getMqttClient().publish(STATIONWARNING, JSON.toJSONString(warningDto).getBytes(), 0, false);
+				System.out.println("发送预警成功: " + JSON.toJSONString(warningDto));
+				logger.info("发送预警成功: " + JSON.toJSONString(warningDto));
+			}
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
@@ -212,16 +233,11 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 		DynamicDetails dynamicDetails = new DynamicDetails(TABNAME, tabContent);
 		List<DynamicDetails> dynamicDetailsList = new ArrayList<>();
 		dynamicDetailsList.add(dynamicDetails);
-		StringBuilder indexKey = new StringBuilder(pointSystem.getStationAbbr()).append("#")
-				.append(pointSystem.getNumber()).append("#").append(pointSystem.getFunctionNum());
+		StringBuilder indexKey = new StringBuilder(pointSystem.getStation()).append("#").append(pointSystem.getNumber())
+				.append("#").append(pointSystem.getFunctionNum());
 		if (pointSystem.getNumber() == null) {
 			indexKey = new StringBuilder(pointSystem.getStationAbbr()).append("#").append(pointSystem.getFunctionNum());
 		}
-		// 如果是升压站
-		if ("SYZ".equals(pointSystem.getNumber())) {
-			indexKey = new StringBuilder(pointSystem.getStationAbbr()).append("#").append(pointSystem.getNumber())
-					.append("#").append(pointSystem.getFunctionNum());
-		}
 
 		QueryWrapper<EquipmentSpecificIndex> indexWrapper = new QueryWrapper<>();
 		indexWrapper.lambda().eq(EquipmentSpecificIndex::getIndexAddress, pointSystem.getAddress());
@@ -232,24 +248,44 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 		if ("遥信".equals(pointSystem.getType())) {
 			indexValue = esi.getEquipmentIndexName();
 		} else if ("遥测".equals(pointSystem.getType())) {
-			if ("XZ".equals(pointSystem.getStationAbbr())) {
-				String json = esi.getValueEnum();
-				JSONArray arr = JSONArray.parseArray(json);
-				List<JSONObject> list = arr.toJavaList(JSONObject.class);
-				Map<String, String> map = list.stream()
-						.collect(Collectors.toMap(i -> i.getString("key"), i -> i.getString("label")));
-				indexValue = map.get(valueLabe);
-			} else if ("TH".equals(pointSystem.getStationAbbr())) {
+			if ("W005".equals(pointSystem.getStation())) {
+//				String json = esi.getValueEnum();
+//				JSONArray arr = JSONArray.parseArray(json);
+//				List<JSONObject> list = arr.toJavaList(JSONObject.class);
+//				Map<String, String> map = list.stream()
+//						.collect(Collectors.toMap(i -> i.getString("key"), i -> i.getString("label")));
+				indexValue = valueLabe;
+			} else if ("P001".equals(pointSystem.getStation())) {
 
 			}
 		} else {
 
 		}
-		WarningDto WarningDto = new WarningDto(indexKey.toString(), indexValue, null,
-				(String) eqdata.get("sourceAttributionDesc"), (String) eqdata.get("sourceAttribution"),
-				dynamicDetailsList, warningObjectCode, time, (String) eqdata.get("kksms"), "equip",
-				getJumpUrlByInfo(warningObjectCode));
-		return WarningDto;
-	}
 
-}
+		// 如果是升压站
+		if ("SYZ".equals(pointSystem.getNumber())) {
+			// 如果开启升压站预警
+			if (warnSYZ) {
+				indexKey = new StringBuilder(pointSystem.getStation()).append("#").append(pointSystem.getNumber())
+						.append("#").append(pointSystem.getFunctionNum());
+			} else {
+				System.out.println("升压站预警不发送: " + warningObjectCode + " , " + indexValue);
+				return null;
+			}
+		}
+
+		String key = "104_warning:" + warningObjectCode + "_" + indexKey.toString() + "_" + indexValue;
+		// 添加缓存机制 在有限的时间内防止一直触发
+		if (redisUtil.get(key) != null&&redisUse) {
+			System.out.println("预警缓存存在,不触发: " + key);
+			return null;
+		} else {
+			WarningDto WarningDto = new WarningDto(indexKey.toString(), indexValue, null,
+					(String) eqdata.get("sourceAttributionDesc"), (String) eqdata.get("sourceAttribution"),
+					dynamicDetailsList, warningObjectCode, time, (String) eqdata.get("kksms"), "equip",
+					getJumpUrlByInfo(warningObjectCode));
+			redisUtil.set(key, indexValue, limitTime);
+			return WarningDto;
+		}
+	}
+}
\ No newline at end of file

amos-boot-data/amos-boot-data-alarm/src/main/resources/application.properties

@@ -36,3 +36,12 @@ kafka.equipment.test=test88888
 power.station.url=http://172.16.4.29:80/prod-api/fdgl/process/DataInterface
 #电站104采集预警
 power.station.warning=104/data/analysis
+
+#104发送告警间隔时间
+warning.wait.time=100
+#104缓存预警过滤
+warning.redis.use=false
+#104缓存预警过滤过期时间单位秒
+warning.redis.limit.time=3600
+#104升压站预警发送
+warning.SYZ=false
\ No newline at end of file