From 6eb8f65d4a69a550945a92cc916d3c63dee739fa Mon Sep 17 00:00:00 2001 From: jinpengyong <jpy123456> Date: Tue, 22 Aug 2023 16:55:42 +0800 Subject: [PATCH] Merge remote-tracking branch 'origin/wb_21' into qa --- screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java | 36 ++++++++++++++++++++++++------------ 1 files changed, 24 insertions(+), 12 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 c67827c..5ee1a09 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,3 +1,4 @@ +/* package com.moral.api.kafka.consumer; import com.moral.api.service.*; @@ -10,6 +11,9 @@ import org.springframework.stereotype.Component; import org.springframework.util.ObjectUtils; +import java.text.DecimalFormat; +import java.text.SimpleDateFormat; +import java.util.Date; import java.util.HashMap; import java.util.Iterator; import java.util.Map; @@ -18,9 +22,8 @@ import com.moral.constant.KafkaConstants; import com.moral.constant.RedisConstants; -/* - * ��������������������� - * */ + + @Component @Slf4j public class DeviceConsumer { @@ -36,9 +39,6 @@ @Autowired private RedisTemplate redisTemplate; - - @Autowired - private HistorySecondSpecialService historySecondSpecialService; @Autowired private HistorySecondCruiserService historySecondCruiserService; @@ -61,7 +61,7 @@ } //������������ - data.remove("time"); +// data.remove("time"); data.remove("entryTime"); Iterator<Map.Entry<String, Object>> iterator = data.entrySet().iterator(); Map<String, Object> newMap = new HashMap<>(); @@ -102,6 +102,17 @@ //������������ data.remove("time"); data.remove("entryTime"); + SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + String format = dateFormat.format(new Date().getTime()); + //������������ + String substring = format.substring(0, 4); + + String substring1 = time.toString().substring(0, 4); + + String replace = time.toString().replace(substring1, substring); + + data.put("DataTime",replace); + Iterator<Map.Entry<String, Object>> iterator = data.entrySet().iterator(); Map<String, Object> newMap = new HashMap<>(); Map.Entry<String, Object> next; @@ -126,7 +137,7 @@ //������������������������������������������������������ @KafkaListener(topics = KafkaConstants.TOPIC_SECOND, containerFactory = "stateListenerContainerFactory") - public void listenSecond(ConsumerRecord<String, String> record) { + public void listenSecond(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { Map<String, Object> data = JSON.parseObject(msg, Map.class); @@ -141,13 +152,14 @@ data.remove("entryTime"); //������������ - data = deviceService.adjustDeviceData(data); + data = deviceService.adjustDeviceData(data,"0"); //������redis data.put("DataTime", time); redisTemplate.opsForHash().put(RedisConstants.DATA_SECOND, mac, data); //��������������������������� data.put("mac", mac); deviceService.judgeDeviceState(data); + ack.acknowledge(); } catch (Exception e) { log.error("param{}" + msg); } @@ -157,7 +169,6 @@ @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"); @@ -180,7 +191,7 @@ } //������������������ - @KafkaListener(topics = KafkaConstants.CRUISER_TOPIC_SECOND, groupId = KafkaConstants.GROUP_INSERT, containerFactory = "insertListenerContainerFactory") + @KafkaListener(topics = KafkaConstants.CRUISER_TOPIC_SECOND, containerFactory = "insertListenerContainerFactory") public void listenSecondCruiser(ConsumerRecord<String, String> record, Acknowledgment ack) { String msg = record.value(); try { @@ -200,7 +211,8 @@ historySecondCruiserService.insertHistorySecond(data); ack.acknowledge(); } catch (Exception e) { - log.error("param{}" + msg); + log.error("param{}" + e); } } } +*/ -- Gitblit v1.8.0