From b888331b6ba56caeeab12a4543fe81bac8854794 Mon Sep 17 00:00:00 2001 From: lizijie <lzjiiie@163.com> Date: Tue, 31 Aug 2021 17:09:46 +0800 Subject: [PATCH] Merge remote-tracking branch 'origin/dev' into dev --- screen-manage/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java | 23 ++++++++--------------- 1 files changed, 8 insertions(+), 15 deletions(-) diff --git a/screen-manage/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java b/screen-manage/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java index 04e47cb..211b9f5 100644 --- a/screen-manage/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java +++ b/screen-manage/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java @@ -24,7 +24,6 @@ /* * ������������������ * */ -@Component @Slf4j public class KafkaConsumer { @@ -44,15 +43,14 @@ private HistorySecondSpecialService historySecondSpecialService; //������������ - @KafkaListener(topics = KafkaConstants.TOPIC_MINUTE, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "kafkaListenerContainerFactory") + @KafkaListener(topics = KafkaConstants.TOPIC_MINUTE, containerFactory = "insertListenerContainerFactory") public void listenMinute(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { Map<String, Object> data = JSON.parseObject(msg, Map.class); Object mac = data.get("mac"); Object time = data.get("DataTime"); - Object ver = data.get("ver"); - if (ObjectUtils.isEmpty(ver) || ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { + if (ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { log.warn("some properties is null, param{}", msg); ack.acknowledge(); return; @@ -84,15 +82,14 @@ } //������������ - @KafkaListener(topics = KafkaConstants.TOPIC_HOUR, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "kafkaListenerContainerFactory") + @KafkaListener(topics = KafkaConstants.TOPIC_HOUR, containerFactory = "insertListenerContainerFactory") public void listenHour(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { Map<String, Object> data = JSON.parseObject(msg, Map.class); Object mac = data.get("mac"); Object time = data.get("DataTime"); - Object ver = data.get("ver"); - if (ObjectUtils.isEmpty(ver) || ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { + if (ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { log.warn("some properties is null, param{}", msg); ack.acknowledge(); return; @@ -124,22 +121,20 @@ } //������������������������������������������������������ - @KafkaListener(topics = KafkaConstants.TOPIC_SECOND, groupId = KafkaConstants.GROUP_STATE, containerFactory = "kafkaListenerContainerFactory") + @KafkaListener(topics = KafkaConstants.TOPIC_SECOND, containerFactory = "stateListenerContainerFactory") public void listenSecond(ConsumerRecord<String, String> record) { String msg = record.value(); try { Map<String, Object> data = JSON.parseObject(msg, Map.class); Object mac = data.get("mac"); Object time = data.get("DataTime"); - Object ver = data.get("ver"); - if (ObjectUtils.isEmpty(ver) || ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { + if (ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { log.warn("some properties is null, param{}", msg); return; } //������������ data.remove("time"); data.remove("entryTime"); - data.remove("ver"); //������������ data = deviceService.adjustDeviceData(data); @@ -153,15 +148,14 @@ } //��������������������� - @KafkaListener(topics = KafkaConstants.TOPIC_SECOND_SPECIAL, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "kafkaListenerContainerFactory") + @KafkaListener(topics = KafkaConstants.UAV_TOPIC_SECOND, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "kafkaListenerContainerFactory") public void listenSecondSpecial(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { Map<String, Object> data = JSON.parseObject(msg, Map.class); Object mac = data.get("mac"); Object time = data.get("DataTime"); - Object ver = data.get("ver"); - if (ObjectUtils.isEmpty(ver) || ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { + if (ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { log.warn("some properties is null, param{}", msg); ack.acknowledge(); return; @@ -170,7 +164,6 @@ //������������ data.remove("time"); data.remove("entryTime"); - data.remove("ver"); historySecondSpecialService.insertHistorySecond(data); ack.acknowledge(); -- Gitblit v1.8.0