修改物联监盘同步数据

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

@@ -21,18 +21,24 @@ public class AlarmKafkaConsumer {
 
 	// 消费者来处理消息
 	@KafkaListener(id = "alarmInfo", topics = { "${kafka.equipment.alarm}" })
-	public void message1(String record, Acknowledgment ack) {
+	public void message1(String record, Acknowledgment ack) throws InterruptedException {
 		// 处理业务
 		String date = record;
+		
 		// 异步触发预警
 		pointSystemServiceImpl.sendWarningAsync(date);
+		
+		// 等待1s
+		Thread.sleep(1000L);
+		
 		// 手动提交
 		ack.acknowledge();
 	}
 
-	@KafkaListener(id = "user2", topics = { "${kafka.equipment.test}" })
+	@KafkaListener(id = "alarmInfo2", topics = { "${kafka.equipment.test}" })
 	public void message2(String record, Acknowledgment ack) {
 		String date = record;
+		System.out.println("收到告警信息" + date);
 	}
 
 //    public void message1( ConsumerRecord<?, ?> record, Acknowledgment ack){

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

@@ -125,14 +125,14 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 
 	@Override
 	public void sendWarning(String address, String value, String valueLabe, String gatewayId, String isAlarm) {
-
 		// 对应 equipment库的wl_equipment_specific_index_alarm_dic表
 		String[] s = { "1", "7", "9" };
 		// 如果不满足择返回
 		if (!Arrays.asList(s).contains(isAlarm)) {
-			//System.out.println("不满足告警类型: " + isAlarm);
+			System.out.println("不满足告警类型: " + isAlarm);
 			return;
 		}
+
 		System.out.println("满足告警消息address: " + address + ",gatewayId: " + gatewayId + " ,value:" + value
 				+ " ,valueLabe: " + valueLabe + " ,isAlarm: " + isAlarm);
 		logger.info("满足告警消息address: " + address + ",gatewayId: " + gatewayId + " ,value:" + value + " ,valueLabe: "
@@ -158,6 +158,9 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 		KKSDataWrapper.lambda().eq(KKSData::getKKSBM, pointSystem.getKks());
 		KKSData KKSData = kksDataMapper.selectOne(KKSDataWrapper);
 		JSONObject eqdata = new JSONObject();
+		if (KKSData == null) {
+			throw new RuntimeException("kks码查询热工院表不存在:" + pointSystem.getKks());
+		}
 		eqdata.put("kksms", KKSData.getKKSMS());
 		
 		QueryWrapper<StationBasic> stationWrapper = new QueryWrapper<>();
@@ -211,6 +214,9 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 		dynamicDetailsList.add(dynamicDetails);
 		StringBuilder indexKey = new StringBuilder(pointSystem.getStationAbbr()).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())
@@ -230,8 +236,9 @@ public class PointSystemServiceImpl extends ServiceImpl<PointSystemMapper, Point
 				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);
+				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())) {
 
 			}

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

@@ -15,10 +15,9 @@ redis.cache.failure.time=10800
 # mybatis-plus
 mybatis-plus.mapper-locations=classpath:mapper/*Mapper.xml
 
-
 #消费者所在组的名称
 #消费者 的broker地址
-spring.kafka.consumer.bootstrap-servers=10.20.0.223:9092
+spring.kafka.consumer.bootstrap-servers=10.20.0.223:9092,10.20.0.133:9092
 # 是否自动提交
 spring.kafka.consumer.enable-auto-commit=false
 #offset的消费位置
@@ -30,26 +29,10 @@ spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.S
 spring.kafka.listener.ack-mode=manual_immediate
 #监听类型
 spring.kafka.listener.type=single
-# 并发
-#spring.kafka.listener.concurrency=5
-
-# 发生错误后，消息重发的次数。
-spring.kafka.producer.retries=1
-#配置kafak produce的broker地址
-spring.kafka.producer.bootstrap-servers=10.20.0.223:9092
-#默认批处理大小（以字节为单位）
-spring.kafka.producer.batch-size=16384
-#生产者可以用来缓冲等待发送到服务器的记录的内存总字节数
-spring.kafka.producer.buffer-memory=33554432
-# producer配置序列化
-spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
-# kafka默认的String序列化器
-spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
 
 kafka.equipment.alarm=EQUIPMENT_ALARM
 kafka.equipment.test=test88888
-
 #电站对接第三方查询设备kks码
-power.station.url=http://10.20.1.151:80/prod-api/fdgl/process/DataInterface
+power.station.url=http://172.16.4.29:80/prod-api/fdgl/process/DataInterface
 #电站104采集预警
 power.station.warning=104/data/analysis