fix(equip):添加告警转发消息至Kafka

amos-boot-module/amos-boot-module-api/amos-boot-module-equip-api/src/main/java/com/yeejoin/equipmanage/common/entity/vo/EquipmentIndexVO.java

@@ -66,6 +66,9 @@ public class EquipmentIndexVO {
     @ApiModelProperty(value = "是否支持趋势查看")
     private Integer isTrend;
 
+    @ApiModelProperty(value = "是否告警")
+    private Integer isAlarm;
+
     @ApiModelProperty(value = "指标枚举")
     private String valueEnum;
 

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/kafka/KafkaConsumerService.java

@@ -10,6 +10,7 @@ import com.yeejoin.equipmanage.common.entity.vo.EquipmentIndexVO;
 import lombok.extern.slf4j.Slf4j;
 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.support.Acknowledgment;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@@ -35,11 +36,19 @@ public class KafkaConsumerService {
     @Autowired
     private InfluxDbConnection influxDbConnection;
 
+    @Autowired
+    protected KafkaProducerService kafkaProducerService;
+
     private Executor dataExecutor = new ThreadPoolTaskExecutor();
 
     @Autowired
     private RedisUtils redisUtils;
 
+    @Value("${kafka.alarm.topic}")
+    private String alarmTopic;
+
+    private static final String MEASUREMENT= "iot_data";
+
     @KafkaListener(topics = "#{'${kafka.topic}'.split(',')}", groupId = "messageConsumerGroup")
     public void listen(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
         try {
@@ -92,17 +101,18 @@ public class KafkaConsumerService {
                         fieldsMap.put("valueLabel", valueLabel.equals("") ? value : valueLabel);
                         fieldsMap.put("gatewayId", gatewayId);
                         fieldsMap.put("dataType", dataType);
-                        fieldsMap.put("equipmentId", equipmentSpeIndex.getEquipmentId());
+                        fieldsMap.put("isAlarm", equipmentSpeIndex.getIsAlarm());
                         fieldsMap.put("equipmentSpecificName", equipmentSpeIndex.getEquipmentSpecificName());
                         fieldsMap.put("equipmentIndexName", equipmentSpeIndex.getEquipmentIndexName());
                         fieldsMap.put("unit", equipmentSpeIndex.getUnitName());
                         SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                         fieldsMap.put("createdTime", simpleDateFormat.format(new Date()));
                         fieldsMap.put("equipmentIndex", JSON.toJSONString(equipmentSpeIndex));
+                        influxDbConnection.insert(MEASUREMENT, tagsMap, fieldsMap);
 
-                        //保存influxDB库
-                        influxDbConnection.insert("iot_data", tagsMap, fieldsMap);
-                        log.info("influxdb入库时间:{}",simpleDateFormat.format(new Date()));
+                        if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
+                            kafkaProducerService.sendMessageAsync(alarmTopic,JSON.toJSONString(fieldsMap));
+                        }
                     }
                 } catch (Exception e) {
                     log.error("Iot透传消息解析入库失败" + e.getMessage(), e);

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/kafka/config/KafkaConsumerConfig.java

@@ -29,6 +29,9 @@ public class KafkaConsumerConfig {
     @Value("${spring.kafka.consumer.group-id}")
     private String kafkaGroupId;
 
+    @Value("${spring.kafka.consumer.enable-auto-commit}")
+    private boolean enableAutoCommit;
+
     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
         factory.setConsumerFactory(consumerFactory());
@@ -49,7 +52,7 @@ public class KafkaConsumerConfig {
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);
 
         // 自动提交
-        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.FALSE);
+        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
         //两次Poll之间的最大允许间隔。
         //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Consumer Group移除并触发Rebalance，默认30s。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);

amos-boot-system-equip/src/main/resources/application-dev.properties

@@ -170,4 +170,5 @@ spring.kafka.listener.ack-mode=manual_immediate
 spring.kafka.listener.type=batch
 kafka.topic=PERSPECTIVE
 emq.topic=iot/data/perspective
+kafka.alarm.topic=EQUIPMENT_ALARM
 iot.async.flag = false
\ No newline at end of file

amos-boot-system-equip/src/main/resources/mapper/EquipmentSpecificIndexMapper.xml

@@ -332,7 +332,8 @@
         si.gateway_id,
         si.data_type,
         si.equipment_specific_name,
-        si.equipment_index_name
+        si.equipment_index_name,
+        si.is_alarm
         FROM
         wl_equipment_specific_index si
         LEFT JOIN wl_equipment_index ei ON si.equipment_index_id = ei.id