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