Skip to content

A
amos-boot-biz
Commit bf88f772 authored by 刘林's avatar 刘林
Browse files
Options

fix(equip):优化对接IOT代码，添加kafka消息队列

parent c014e982
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 361 additions and 355 deletions
amos-boot-data/amos-boot-data-equip/pom.xml
View file @ bf88f772
...@@ -28,11 +28,11 @@ ...@@ -28,11 +28,11 @@
<artifactId>spring-kafka</artifactId> <artifactId>spring-kafka</artifactId>
</dependency> </dependency>
<!-- <dependency>--> <dependency>
<!-- <groupId>org.typroject</groupId>--> <groupId>org.typroject</groupId>
<!-- <artifactId>tyboot-component-emq</artifactId>--> <artifactId>tyboot-component-emq</artifactId>
<!-- <version>1.1.20</version>--> <version>1.1.23</version>
<!-- </dependency>--> </dependency>
<dependency> <dependency>
<groupId>com.yeejoin</groupId> <groupId>com.yeejoin</groupId>
......
amos-boot-data/amos-boot-data-equip/src/main/java/com/yeejoin/equip/eqmx/EmqMessageService.java
View file @ bf88f772
//package com.yeejoin.equip.eqmx; package com.yeejoin.equip.eqmx;
//
//import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
//import com.yeejoin.equip.kafka.KafkaProducerService; import com.yeejoin.equip.kafka.KafkaProducerService;
//import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
//import net.sf.json.JSONObject; import net.sf.json.JSONObject;
//import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttMessage;
//import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
//import org.typroject.tyboot.component.emq.EmqKeeper; import org.typroject.tyboot.component.emq.EmqKeeper;
//import org.typroject.tyboot.component.emq.EmqxListener; import org.typroject.tyboot.component.emq.EmqxListener;
//import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
//import java.util.Arrays; import java.util.Arrays;
//import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
//import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
//
///** /**
// * @author LiuLin * @author LiuLin
// * @date 2023/6/25 * @date 2023/6/25
// * @apiNote Emq消息转发Kafka * @apiNote Emq消息转发Kafka
// */ */
//@Slf4j @Slf4j
//@Component @Component
//public class EmqMessageService extends EmqxListener { public class EmqMessageService extends EmqxListener {
//
// @Autowired @Autowired
// protected EmqKeeper emqKeeper; protected EmqKeeper emqKeeper;
//
// @Autowired @Autowired
// protected KafkaProducerService kafkaProducerService; protected KafkaProducerService kafkaProducerService;
//
// @Value("${emq.topic}") @Value("${emq.topic}")
// private String emqTopic; private String emqTopic;
//
// @Value("${kafka.topic}") @Value("${kafka.topic}")
// private String kafkaTopic; private String kafkaTopic;
//
// private static final BlockingQueue<JSONObject> blockingQueue = new LinkedBlockingQueue<>(); private static final BlockingQueue<JSONObject> blockingQueue = new LinkedBlockingQueue<>();
//
// @PostConstruct @PostConstruct
// void init() throws Exception { void init() {
// emqKeeper.subscript(emqTopic, 1, this); new Thread(task_runnable).start();
// } String[] split = emqTopic.split(",");
// Arrays.stream(split).forEach(e-> {
// @Override try {
// public void processMessage(String topic, MqttMessage message) throws Exception { emqKeeper.subscript(e, 1, this);
// JSONObject result = JSONObject.fromObject(new String(message.getPayload())); } catch (Exception exception) {
// //JSONObject messageResult = new JSONObject(); log.info("订阅emq消息失败 ====> message: {}", exception.getMessage());
// //messageResult.put("result", result); }
// //messageResult.put("topic", topic); });
// //blockingQueue.add(messageResult); }
//
// if (topic.equals(emqTopic)) { @Override
// kafkaProducerService.sendMessageAsync(kafkaTopic,JSON.toJSONString(result)); public void processMessage(String topic, MqttMessage message) throws Exception {
// } JSONObject result = JSONObject.fromObject(new String(message.getPayload()));
// } JSONObject messageResult = new JSONObject();
// messageResult.put("result", result);
// //Runnable task_runnable = new Runnable() { messageResult.put("topic", topic);
// // public void run() { blockingQueue.add(messageResult);
// // int k = 0; }
// // boolean b = true;
// // while (b) { Runnable task_runnable = new Runnable() {
// // k++; public void run() {
// // b = k < Integer.MAX_VALUE; int k = 0;
// // try { boolean b = true;
// // JSONObject messageResult = blockingQueue.take(); while (b) {
// // JSONObject result = messageResult.getJSONObject("result"); k++;
// // if ((messageResult.getString("topic")).equals(emqTopic)) { b = k < Integer.MAX_VALUE;
// // kafkaProducerService.sendMessageAsync(kafkaTopic,JSON.toJSONString(result)); try {
// // } JSONObject messageResult = blockingQueue.take();
// // } catch (Exception e) { JSONObject result = messageResult.getJSONObject("result");
// // Thread.currentThread().interrupt(); if ((messageResult.getString("topic")).equals(emqTopic)) {
// // } kafkaProducerService.sendMessageAsync(kafkaTopic,JSON.toJSONString(result));
// // } }
// // } } catch (Exception e) {
// //}; Thread.currentThread().interrupt();
//} }
}
}
};
}
amos-boot-data/amos-boot-data-equip/src/main/java/com/yeejoin/equip/mqtt/MessageIntegration.java
View file @ bf88f772
package com.yeejoin.equip.mqtt; //package com.yeejoin.equip.mqtt;
//
import com.yeejoin.equip.config.MqttPropertyConfig; //import com.yeejoin.equip.config.MqttPropertyConfig;
import com.yeejoin.equip.entity.IndicatorData; //import com.yeejoin.equip.entity.IndicatorData;
import com.yeejoin.equip.mqtt.message.MqttTopicEnum; //import com.yeejoin.equip.mqtt.message.MqttTopicEnum;
import com.yeejoin.equip.utils.ExecutorFactory; //import com.yeejoin.equip.utils.ExecutorFactory;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions; //import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.context.annotation.Bean; //import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; //import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator; //import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel; //import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow; //import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows; //import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.endpoint.MessageProducerSupport; //import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; //import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory; //import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; //import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; //import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; //import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel; //import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler; //import org.springframework.messaging.MessageHandler;
import javax.annotation.Resource; //import javax.annotation.Resource;
import java.util.Objects; //import java.util.Objects;
import static com.yeejoin.equip.mqtt.message.MqttConstant.*; //import static com.yeejoin.equip.mqtt.message.MqttConstant.*;
//
/** ///**
* 消息处理器 // * 消息处理器
* // *
* @author LiuLin // * @author LiuLin
* @date 2023年08月18日 10:56 // * @date 2023年08月18日 10:56
*/ // */
@Configuration //@Configuration
public class MessageIntegration { //public class MessageIntegration {
//
@Resource // @Resource
private MqttPropertyConfig mqttPropertyConfig; // private MqttPropertyConfig mqttPropertyConfig;
@Bean // @Bean
public MqttConnectOptions mqttConnectOptions() { // public MqttConnectOptions mqttConnectOptions() {
MqttConnectOptions options = new MqttConnectOptions(); // MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[]{mqttPropertyConfig.getBroker()}); // options.setServerURIs(new String[]{mqttPropertyConfig.getBroker()});
options.setUserName(mqttPropertyConfig.getClientUserName()); // options.setUserName(mqttPropertyConfig.getClientUserName());
options.setPassword(mqttPropertyConfig.getClientPassword().toCharArray()); // options.setPassword(mqttPropertyConfig.getClientPassword().toCharArray());
options.setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT); // options.setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT);
// 设置心跳：1.5*20秒 // // 设置心跳：1.5*20秒
options.setKeepAliveInterval(mqttPropertyConfig.getKeepAliveInterval()); // options.setKeepAliveInterval(mqttPropertyConfig.getKeepAliveInterval());
// 设置最大并发数 // // 设置最大并发数
options.setMaxInflight(mqttPropertyConfig.getMaxInflight()); // options.setMaxInflight(mqttPropertyConfig.getMaxInflight());
options.setAutomaticReconnect(true); // options.setAutomaticReconnect(true);
//options.setCleanSession(false); // //options.setCleanSession(false);
return options; // return options;
} // }
//
@Bean // @Bean
public MqttPahoClientFactory mqttClientFactory() { // public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); // DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(mqttConnectOptions()); // factory.setConnectionOptions(mqttConnectOptions());
return factory; // return factory;
} // }
//
@Bean // @Bean
public MessageProducerSupport bizInbound() { // public MessageProducerSupport bizInbound() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter( // MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
mqttPropertyConfig.getBizClientId(), // mqttPropertyConfig.getBizClientId(),
mqttClientFactory(), // mqttClientFactory(),
mqttPropertyConfig.getBizTopic() // mqttPropertyConfig.getBizTopic()
); // );
adapter.setCompletionTimeout(DEFAULT_COMPLETION_TIMEOUT); // adapter.setCompletionTimeout(DEFAULT_COMPLETION_TIMEOUT);
adapter.setConverter(new DefaultPahoMessageConverter()); // adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(QOS_DEFAULT); // adapter.setQos(QOS_DEFAULT);
return adapter; // return adapter;
} // }
//
@Bean // @Bean
public MessageChannel mqttOutboundChannel() { // public MessageChannel mqttOutboundChannel() {
return new DirectChannel(); // return new DirectChannel();
} // }
//
@Bean // @Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel") // @ServiceActivator(inputChannel = "mqttOutboundChannel")
public MessageHandler mqttOutbound() { // public MessageHandler mqttOutbound() {
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler( // MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
mqttPropertyConfig.getClientId(), // mqttPropertyConfig.getClientId(),
mqttClientFactory() // mqttClientFactory()
); // );
messageHandler.setAsync(true); // messageHandler.setAsync(true);
messageHandler.setDefaultQos(QOS_DEFAULT); // messageHandler.setDefaultQos(QOS_DEFAULT);
return messageHandler; // return messageHandler;
} // }
//
@Bean // @Bean
public IntegrationFlow bizMsgFlow() { // public IntegrationFlow bizMsgFlow() {
return IntegrationFlows // return IntegrationFlows
.from(bizInbound()) // .from(bizInbound())
.channel(channels -> channels.executor(ExecutorFactory.buildBizExecutor())) // .channel(channels -> channels.executor(ExecutorFactory.buildBizExecutor()))
.handle(MessageTransfer::mqttMessage2RawMessage) // .handle(MessageTransfer::mqttMessage2RawMessage)
//根据Topic后缀进行分流 // //根据Topic后缀进行分流
.<IndicatorData, MqttTopicEnum>route(IndicatorData::getMqttTopicEnum, // .<IndicatorData, MqttTopicEnum>route(IndicatorData::getMqttTopicEnum,
mapping -> mapping // mapping -> mapping
.subFlowMapping(MqttTopicEnum.perspective, flow -> flow // .subFlowMapping(MqttTopicEnum.perspective, flow -> flow
.handle("handleMessageService", "processMessage") // .handle("handleMessageService", "processMessage")
.filter(Objects::nonNull) // .filter(Objects::nonNull)
.handle(mqttOutbound())) // .handle(mqttOutbound()))
.defaultOutputToParentFlow()) // .defaultOutputToParentFlow())
.get(); // .get();
} // }
} //}
amos-boot-data/amos-boot-data-equip/src/main/java/com/yeejoin/equip/mqtt/MessageTransfer.java
View file @ bf88f772
package com.yeejoin.equip.mqtt; //package com.yeejoin.equip.mqtt;
//
import com.alibaba.fastjson.JSON; //import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray; //import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject; //import com.alibaba.fastjson.JSONObject;
import com.yeejoin.equip.entity.EquipmentIndexVO; //import com.yeejoin.equip.entity.EquipmentIndexVO;
import com.yeejoin.equip.entity.IndicatorData; //import com.yeejoin.equip.entity.IndicatorData;
import com.yeejoin.equip.mqtt.message.MqttTopicEnum; //import com.yeejoin.equip.mqtt.message.MqttTopicEnum;
import com.yeejoin.equip.utils.RedisUtils; //import com.yeejoin.equip.utils.RedisUtils;
import com.yeejoin.equip.utils.SpringUtils; //import com.yeejoin.equip.utils.SpringUtils;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils; //import org.apache.commons.lang3.ObjectUtils;
import org.springframework.integration.mqtt.support.MqttHeaders; //import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
import java.util.Arrays; //import java.util.Arrays;
import java.util.Date; //import java.util.Date;
import java.util.Map; //import java.util.Map;
import static com.yeejoin.equip.mqtt.message.MqttConstant.*; //import static com.yeejoin.equip.mqtt.message.MqttConstant.*;
//
/** ///**
* @author LiuLin // * @author LiuLin
* @date 2023年07月13日 09:58 // * @date 2023年07月13日 09:58
*/ // */
@Slf4j //@Slf4j
@Component //@Component
public class MessageTransfer { //public class MessageTransfer {
//
/** // /**
* 转为原生数据，payload为字节数组 // * 转为原生数据，payload为字节数组
**/ // **/
public static IndicatorData mqttMessage2RawMessage(String payload, Map<String, Object> headers) { // public static IndicatorData mqttMessage2RawMessage(String payload, Map<String, Object> headers) {
//log.info("received raw message, header >>> {}, payload >>> {}", headers, JSONObject.toJSONString(payload)); // //log.info("received raw message, header >>> {}, payload >>> {}", headers, JSONObject.toJSONString(payload));
RedisUtils redisUtils = (RedisUtils) SpringUtils.getBean("redisUtils"); // RedisUtils redisUtils = (RedisUtils) SpringUtils.getBean("redisUtils");
IndicatorData indicatorData = JSON.parseObject(payload, IndicatorData.class); // IndicatorData indicatorData = JSON.parseObject(payload, IndicatorData.class);
if(!indicatorData.getSignalType().equals("transformation")){ // if(!indicatorData.getSignalType().equals("transformation")){
log.info("received TotalSummon message:address{},gateway:{},time:{},signalType:{}", indicatorData.getAddress(),indicatorData.getGatewayId(), // log.info("received TotalSummon message:address{},gateway:{},time:{},signalType:{}", indicatorData.getAddress(),indicatorData.getGatewayId(),
new Date(),indicatorData.getSignalType()); // new Date(),indicatorData.getSignalType());
} // }
//log.info("received raw message, header >>> {}, payload >>> {}", headers, JSONObject.toJSONString(payload)); // //log.info("received raw message, header >>> {}, payload >>> {}", headers, JSONObject.toJSONString(payload));
try { // try {
String topic = headers.get(MqttHeaders.RECEIVED_TOPIC).toString(); // String topic = headers.get(MqttHeaders.RECEIVED_TOPIC).toString();
String[] topicItems = topic.split(TOPIC_SPLITTER); // String[] topicItems = topic.split(TOPIC_SPLITTER);
indicatorData.setMqttTopicEnum(MqttTopicEnum.of(topicItems[topicItems.length - 1])); // indicatorData.setMqttTopicEnum(MqttTopicEnum.of(topicItems[topicItems.length - 1]));
String key = indicatorData.getAddress() + "_" + indicatorData.getGatewayId(); // String key = indicatorData.getAddress() + "_" + indicatorData.getGatewayId();
if (redisUtils.hasKey(key)) { // if (redisUtils.hasKey(key)) {
EquipmentIndexVO equipmentSpeIndex = JSONObject.parseObject(redisUtils.get(key),EquipmentIndexVO.class) ; // EquipmentIndexVO equipmentSpeIndex = JSONObject.parseObject(redisUtils.get(key),EquipmentIndexVO.class) ;
String valueLabel = valueTranslate(indicatorData.getValue(), equipmentSpeIndex.getValueEnum()); // String valueLabel = valueTranslate(indicatorData.getValue(), equipmentSpeIndex.getValueEnum());
indicatorData.setIsAlarm(String.valueOf(equipmentSpeIndex.getIsAlarm())); // indicatorData.setIsAlarm(String.valueOf(equipmentSpeIndex.getIsAlarm()));
indicatorData.setEquipmentIndexName(equipmentSpeIndex.getEquipmentIndexName()); // indicatorData.setEquipmentIndexName(equipmentSpeIndex.getEquipmentIndexName());
indicatorData.setEquipmentSpecificName(equipmentSpeIndex.getEquipmentSpecificName()); // indicatorData.setEquipmentSpecificName(equipmentSpeIndex.getEquipmentSpecificName());
indicatorData.setUnit(equipmentSpeIndex.getUnitName()); // indicatorData.setUnit(equipmentSpeIndex.getUnitName());
indicatorData.setEquipmentsIdx(key); // indicatorData.setEquipmentsIdx(key);
indicatorData.setValueLabel(valueLabel.isEmpty() ? indicatorData.getValue() : valueLabel); // indicatorData.setValueLabel(valueLabel.isEmpty() ? indicatorData.getValue() : valueLabel);
if (!Arrays.asList(TRUE, FALSE).contains(indicatorData.getValue())) { // if (!Arrays.asList(TRUE, FALSE).contains(indicatorData.getValue())) {
indicatorData.setValueF(Float.parseFloat(indicatorData.getValue())); // indicatorData.setValueF(Float.parseFloat(indicatorData.getValue()));
} // }
}else { // }else {
return null; // return null;
} // }
} catch (Exception e) { // } catch (Exception e) {
log.error("mqttMessage2RawMessage解析消息数据异常", e); // log.error("mqttMessage2RawMessage解析消息数据异常", e);
} // }
return indicatorData; // return indicatorData;
} // }
//
private static String valueTranslate(String value, String enumStr) { // private static String valueTranslate(String value, String enumStr) {
if (ObjectUtils.isEmpty(enumStr)) { // if (ObjectUtils.isEmpty(enumStr)) {
return ""; // return "";
} // }
try { // try {
JSONArray jsonArray = JSONArray.parseArray(enumStr); // JSONArray jsonArray = JSONArray.parseArray(enumStr);
for (int i = 0; i < jsonArray.size(); i++) { // for (int i = 0; i < jsonArray.size(); i++) {
JSONObject jsonObject = jsonArray.getJSONObject(i); // JSONObject jsonObject = jsonArray.getJSONObject(i);
if (jsonObject.get("key").equals(value)) { // if (jsonObject.get("key").equals(value)) {
return jsonObject.getString("label"); // return jsonObject.getString("label");
} // }
} // }
} catch (Exception e) { // } catch (Exception e) {
log.error("告警枚举转换异常" + e.getMessage(), e); // log.error("告警枚举转换异常" + e.getMessage(), e);
} // }
return ""; // return "";
} // }
} //}
amos-boot-data/amos-boot-data-equip/src/main/java/com/yeejoin/equip/service/HandleMessageService.java
View file @ bf88f772
package com.yeejoin.equip.service; //package com.yeejoin.equip.service;
//
import com.alibaba.fastjson.JSON; //import com.alibaba.fastjson.JSON;
import com.yeejoin.amos.component.influxdb.InfluxDbConnection; //import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
import com.yeejoin.equip.entity.IndicatorData; //import com.yeejoin.equip.entity.IndicatorData;
import com.yeejoin.equip.kafka.KafkaProducerService; //import com.yeejoin.equip.kafka.KafkaProducerService;
import com.yeejoin.equip.mapper.tdengine.IndicatorDataMapper; //import com.yeejoin.equip.mapper.tdengine.IndicatorDataMapper;
import com.yeejoin.equip.utils.ElasticSearchUtil; //import com.yeejoin.equip.utils.ElasticSearchUtil;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value; //import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat; //import java.text.SimpleDateFormat;
import java.util.Date; //import java.util.Date;
import java.util.HashMap; //import java.util.HashMap;
import java.util.Map; //import java.util.Map;
/** ///**
* @author LiuLin // * @author LiuLin
* @date 2023/6/25 // * @date 2023/6/25
* @apiNote Emq消息转发Kafka // * @apiNote Emq消息转发Kafka
*/ // */
@Slf4j //@Slf4j
@Component("handleMessageService") //@Component("handleMessageService")
public class HandleMessageService { //public class HandleMessageService {
private static final String MEASUREMENT = "iot_data_"; // private static final String MEASUREMENT = "iot_data_";
private static final String TOTAL_DATA_ = "total_data_"; // private static final String TOTAL_DATA_ = "total_data_";
private static final String ES_INDEX_NAME_JX = "jxiop_equipments"; // private static final String ES_INDEX_NAME_JX = "jxiop_equipments";
@Autowired // @Autowired
protected KafkaProducerService kafkaProducerService; // protected KafkaProducerService kafkaProducerService;
@Autowired // @Autowired
private InfluxDbConnection influxDbConnection; // private InfluxDbConnection influxDbConnection;
@Autowired // @Autowired
private IndicatorDataMapper indicatorDataMapper; // private IndicatorDataMapper indicatorDataMapper;
@Value("${kafka.alarm.topic}") // @Value("${kafka.alarm.topic}")
private String alarmTopic; // private String alarmTopic;
@Autowired // @Autowired
private ElasticSearchUtil elasticSearchUtil; // private ElasticSearchUtil elasticSearchUtil;
//
public void processMessage(IndicatorData indicatorData) { // public void processMessage(IndicatorData indicatorData) {
try { // try {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); // SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Map<String, String> tagsMap = new HashMap<>(); // Map<String, String> tagsMap = new HashMap<>();
Map<String, Object> fieldsMap = new HashMap<>(); // Map<String, Object> fieldsMap = new HashMap<>();
tagsMap.put("equipmentsIdx", indicatorData.getEquipmentsIdx()); // tagsMap.put("equipmentsIdx", indicatorData.getEquipmentsIdx());
//
fieldsMap.put("address", indicatorData.getAddress()); // fieldsMap.put("address", indicatorData.getAddress());
fieldsMap.put("gatewayId", indicatorData.getGatewayId()); // fieldsMap.put("gatewayId", indicatorData.getGatewayId());
fieldsMap.put("dataType", indicatorData.getDataType()); // fieldsMap.put("dataType", indicatorData.getDataType());
fieldsMap.put("isAlarm", indicatorData.getIsAlarm()); // fieldsMap.put("isAlarm", indicatorData.getIsAlarm());
fieldsMap.put("equipmentSpecificName", indicatorData.getEquipmentSpecificName()); // fieldsMap.put("equipmentSpecificName", indicatorData.getEquipmentSpecificName());
fieldsMap.put("value", indicatorData.getValue()); // fieldsMap.put("value", indicatorData.getValue());
fieldsMap.put("valueLabel", indicatorData.getValueLabel()); // fieldsMap.put("valueLabel", indicatorData.getValueLabel());
fieldsMap.put("equipmentIndexName", indicatorData.getEquipmentIndexName()); // fieldsMap.put("equipmentIndexName", indicatorData.getEquipmentIndexName());
fieldsMap.put("unit", indicatorData.getUnit()); // fieldsMap.put("unit", indicatorData.getUnit());
fieldsMap.put("createdTime", simpleDateFormat.format(new Date())); // fieldsMap.put("createdTime", simpleDateFormat.format(new Date()));
//
//更新数据入ES库 // //更新数据入ES库
Map<String, Object> paramJson = new HashMap<>(); // Map<String, Object> paramJson = new HashMap<>();
paramJson.put("valueF", indicatorData.getValueF()); // paramJson.put("valueF", indicatorData.getValueF());
paramJson.put("value", indicatorData.getValue()); // paramJson.put("value", indicatorData.getValue());
paramJson.put("valueLabel", indicatorData.getValueLabel()); // paramJson.put("valueLabel", indicatorData.getValueLabel());
paramJson.put("createdTime", new Date()); // paramJson.put("createdTime", new Date());
paramJson.put("unit", indicatorData.getUnit()); // paramJson.put("unit", indicatorData.getUnit());
elasticSearchUtil.updateData(ES_INDEX_NAME_JX, indicatorData.getEquipmentsIdx(), JSON.toJSONString(paramJson)); // elasticSearchUtil.updateData(ES_INDEX_NAME_JX, indicatorData.getEquipmentsIdx(), JSON.toJSONString(paramJson));
//
if (indicatorData.getIsAlarm() != null && "1".equals(indicatorData.getIsAlarm())) { // if (indicatorData.getIsAlarm() != null && "1".equals(indicatorData.getIsAlarm())) {
fieldsMap.putAll(tagsMap); // fieldsMap.putAll(tagsMap);
kafkaProducerService.sendMessageAsync(alarmTopic, JSON.toJSONString(fieldsMap)); // kafkaProducerService.sendMessageAsync(alarmTopic, JSON.toJSONString(fieldsMap));
} // }
//
if ("transformation".equals(indicatorData.getSignalType())) { // if ("transformation".equals(indicatorData.getSignalType())) {
influxDbConnection.insert(MEASUREMENT + indicatorData.getGatewayId(), tagsMap, fieldsMap); // influxDbConnection.insert(MEASUREMENT + indicatorData.getGatewayId(), tagsMap, fieldsMap);
indicatorDataMapper.insert(indicatorData); // indicatorDataMapper.insert(indicatorData);
log.info("TDEngine入库成功，{},value:{}",indicatorData.getEquipmentsIdx(),indicatorData.getValue()); // log.info("TDEngine入库成功，{},value:{}",indicatorData.getEquipmentsIdx(),indicatorData.getValue());
}else{ // }else{
influxDbConnection.insert(TOTAL_DATA_ + indicatorData.getGatewayId(), tagsMap, fieldsMap); // influxDbConnection.insert(TOTAL_DATA_ + indicatorData.getGatewayId(), tagsMap, fieldsMap);
} // }
//
} catch (Exception e) { // } catch (Exception e) {
log.error("Iot透传消息解析入库失败" + e.getMessage(), e); // log.error("Iot透传消息解析入库失败" + e.getMessage(), e);
} // }
} // }
} //}
amos-boot-data/amos-boot-data-equip/src/main/resources/application-dev.properties
View file @ bf88f772
...@@ -82,6 +82,8 @@ spring.kafka.consumer.max-poll-records=1000 ...@@ -82,6 +82,8 @@ spring.kafka.consumer.max-poll-records=1000
spring.kafka.listener.ack-mode=manual_immediate spring.kafka.listener.ack-mode=manual_immediate
spring.kafka.listener.type=batch spring.kafka.listener.type=batch
kafka.alarm.topic=EQUIPMENT_ALARM kafka.alarm.topic=EQUIPMENT_ALARM
kafka.topic=PERSPECTIVE
emq.topic=iot/data/perspective
elasticsearch.address= 139.9.173.44:9200 elasticsearch.address= 139.9.173.44:9200
elasticsearch.username= elastic elasticsearch.username= elastic
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment