From 659b91b554ba5e98b6fffa045231f5aa3fb4ab9e Mon Sep 17 00:00:00 2001 From: lizijie <lzjiiie@163.com> Date: Fri, 03 Sep 2021 15:21:48 +0800 Subject: [PATCH] 因子默认单位添加;角色、菜单更新 --- screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java | 47 ++++++++++++++++++++++++++++++++++++++--------- 1 files changed, 38 insertions(+), 9 deletions(-) diff --git a/screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java b/screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java index 9e7347a..8156f7f 100644 --- a/screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java +++ b/screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java @@ -1,5 +1,6 @@ package com.moral.api.kafka.consumer; +import com.moral.api.service.*; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.beans.factory.annotation.Autowired; @@ -14,10 +15,6 @@ import java.util.Map; import com.alibaba.fastjson.JSON; -import com.moral.api.service.DeviceService; -import com.moral.api.service.HistoryHourlyService; -import com.moral.api.service.HistoryMinutelyService; -import com.moral.api.service.HistorySecondSpecialService; import com.moral.constant.KafkaConstants; import com.moral.constant.RedisConstants; @@ -43,8 +40,14 @@ @Autowired private HistorySecondSpecialService historySecondSpecialService; + @Autowired + private HistorySecondCruiserService historySecondCruiserService; + + @Autowired + private HistorySecondUavService historySecondUavService; + //������������ - @KafkaListener(topics = KafkaConstants.TOPIC_MINUTE, containerFactory = "insertListenerContainerFactory") + //@KafkaListener(topics = KafkaConstants.TOPIC_MINUTE, containerFactory = "insertListenerContainerFactory") public void listenMinute(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { @@ -83,7 +86,7 @@ } //������������ - @KafkaListener(topics = KafkaConstants.TOPIC_HOUR, containerFactory = "insertListenerContainerFactory") + //@KafkaListener(topics = KafkaConstants.TOPIC_HOUR, containerFactory = "insertListenerContainerFactory") public void listenHour(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { @@ -122,7 +125,7 @@ } //������������������������������������������������������ - @KafkaListener(topics = KafkaConstants.TOPIC_SECOND, containerFactory = "stateListenerContainerFactory") + //@KafkaListener(topics = KafkaConstants.TOPIC_SECOND, containerFactory = "stateListenerContainerFactory") public void listenSecond(ConsumerRecord<String, String> record) { String msg = record.value(); try { @@ -150,9 +153,35 @@ } } - //��������������������� + //������������������ @KafkaListener(topics = KafkaConstants.UAV_TOPIC_SECOND, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "insertListenerContainerFactory") public void listenSecondSpecial(ConsumerRecord<String, String> record, Acknowledgment ack) { + String msg = record.value(); + System.out.println(msg); + try { + Map<String, Object> data = JSON.parseObject(msg, Map.class); + Object mac = data.get("mac"); + Object time = data.get("DataTime"); + if (ObjectUtils.isEmpty(time) || ObjectUtils.isEmpty(mac)) { + log.warn("some properties is null, param{}", msg); + ack.acknowledge(); + return; + } + + //������������ + data.remove("time"); + data.remove("entryTime"); + + historySecondUavService.insertHistorySecond(data); + ack.acknowledge(); + } catch (Exception e) { + log.error("param{}" + msg); + } + } + + //������������������ + @KafkaListener(topics = KafkaConstants.CRUISER_TOPIC_SECOND, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "insertListenerContainerFactory") + public void listenSecondCruiser(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { Map<String, Object> data = JSON.parseObject(msg, Map.class); @@ -168,7 +197,7 @@ data.remove("time"); data.remove("entryTime"); - historySecondSpecialService.insertHistorySecond(data); + historySecondCruiserService.insertHistorySecond(data); ack.acknowledge(); } catch (Exception e) { log.error("param{}" + msg); -- Gitblit v1.8.0