From 5242d687d4edcd6c0f7e10fb82b0e5505dbb3da3 Mon Sep 17 00:00:00 2001 From: jinpengyong <jpy123456> Date: Fri, 25 Aug 2023 08:24:05 +0800 Subject: [PATCH] Merge remote-tracking branch 'origin/wb' into qa --- screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java | 68 +++++++++++++++++++++++++++------ 1 files changed, 55 insertions(+), 13 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 2805587..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,5 +1,7 @@ +/* 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; @@ -9,21 +11,19 @@ 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; 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; -/* - * ��������������������� - * */ + + @Component @Slf4j public class DeviceConsumer { @@ -41,7 +41,10 @@ private RedisTemplate redisTemplate; @Autowired - private HistorySecondSpecialService historySecondSpecialService; + private HistorySecondCruiserService historySecondCruiserService; + + @Autowired + private HistorySecondUavService historySecondUavService; //������������ @KafkaListener(topics = KafkaConstants.TOPIC_MINUTE, containerFactory = "insertListenerContainerFactory") @@ -58,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<>(); @@ -99,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; @@ -123,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); @@ -138,18 +152,20 @@ 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); } } - //��������������������� + //������������������ @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(); @@ -167,10 +183,36 @@ data.remove("time"); data.remove("entryTime"); - historySecondSpecialService.insertHistorySecond(data); + historySecondUavService.insertHistorySecond(data); ack.acknowledge(); } catch (Exception e) { log.error("param{}" + msg); } } + + //������������������ + @KafkaListener(topics = KafkaConstants.CRUISER_TOPIC_SECOND, 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); + 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"); + + historySecondCruiserService.insertHistorySecond(data); + ack.acknowledge(); + } catch (Exception e) { + log.error("param{}" + e); + } + } } +*/ -- Gitblit v1.8.0