From 4e250e6a954ab426cc54dcf11caec806eb8df688 Mon Sep 17 00:00:00 2001 From: cjl <276999030@qq.com> Date: Wed, 20 Sep 2023 17:21:02 +0800 Subject: [PATCH] feat:添加测试更新guid --- screen-manage/src/main/java/com/moral/api/kafka/consumer/DeviceConsumer.java | 41 ++++++++++++++++++++++++++++++++--------- 1 files changed, 32 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 4c8f423..3d3378d 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.alibaba.fastjson.JSONObject; import com.moral.api.service.*; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -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 { @@ -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<>(); @@ -97,8 +100,27 @@ } //������������ - data.remove("time"); +// data.remove("time"); data.remove("entryTime"); + Map<String, Object> deviceByMac = deviceService.getDeviceByMac(mac.toString()); + HashMap<String, Object> map = (HashMap<String, Object>) deviceByMac.get("organization"); + String id = map.get("id").toString(); + if (id.equals("71")){ + log.warn(id, msg); + ack.acknowledge(); + return; + } +// 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 +145,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,13 +160,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); } @@ -176,7 +199,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 { @@ -196,7 +219,7 @@ historySecondCruiserService.insertHistorySecond(data); ack.acknowledge(); } catch (Exception e) { - log.error("param{}" + msg); + log.error("param{}" + e); } } } -- Gitblit v1.8.0