From bc85f620ec6bc7bdbe4c95fc227e1ee71119a9b9 Mon Sep 17 00:00:00 2001
From: jinpengyong <jpy123456>
Date: Tue, 02 Nov 2021 10:00:57 +0800
Subject: [PATCH] 省内排名,日综指计算任务

---
 screen-api/src/main/java/com/moral/api/kafka/consumer/SecondDataConsumer.java |   55 +++++++++++++++++++++++++++++++++++++++----------------
 1 files changed, 39 insertions(+), 16 deletions(-)

diff --git a/screen-api/src/main/java/com/moral/api/kafka/consumer/SecondDataConsumer.java b/screen-api/src/main/java/com/moral/api/kafka/consumer/SecondDataConsumer.java
index 9439f3a..68fcd33 100644
--- a/screen-api/src/main/java/com/moral/api/kafka/consumer/SecondDataConsumer.java
+++ b/screen-api/src/main/java/com/moral/api/kafka/consumer/SecondDataConsumer.java
@@ -6,14 +6,19 @@
 import com.moral.api.entity.UnitConversion;
 import com.moral.api.websocket.SingleDeviceServer;
 import com.moral.util.UnitConvertUtils;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.listener.ConsumerSeekAware;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ObjectUtils;
 
+import java.math.BigDecimal;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -27,23 +32,23 @@
  * @Version TODO
  **/
 @Component
+@Slf4j
 public class SecondDataConsumer implements ConsumerSeekAware {
 
-    @KafkaListener(topics = "second_data", groupId = "SecondsDataGroup996")
+    @KafkaListener(containerFactory = "secondDataListenerFactory", topics = "second_data")
     public void listen(ConsumerRecord<String, String> record, Consumer consumer) throws Exception {
         String messageStr = record.value();
         Map<String, Object> message = (Map<String, Object>) JSON.parse(messageStr);
-        long serverStartTime = System.currentTimeMillis();
+        //long serverStartTime = System.currentTimeMillis();
         CopyOnWriteArraySet<SingleDeviceServer> sockets = SingleDeviceServer.sockets;
         for (SingleDeviceServer socket : sockets) {
-            //���������������������������socket
+            //���������������������������socket
             String messageMac = (String) message.get("mac");
             String mac = socket.getMac();
-            if(!mac.equalsIgnoreCase(messageMac))
+            if (!mac.equalsIgnoreCase(messageMac))
                 continue;
             //������������������
-            Map<String, Device> devicesInfo = socket.getDevicesInfo();
-            Device device = devicesInfo.get(mac);
+            Device device = socket.getDeviceAlarmInfo();
             //������������������������
             Map<String, Object> resultMessgae = new HashMap<>();
             //������������
@@ -54,7 +59,23 @@
                 String showUnitKey = sensor.getShowUnitKey();
                 String unitKey = sensor.getUnitKey();
                 String unit = sensor.getUnit();
-                String sourceData = String.valueOf(message.get(code));
+                //���������������������������������������������
+                Object value = message.get(code);
+                //������������������������������������������������
+                /*if(value==null)    ���������
+                    continue;
+                Double sourceDataD = Double.valueOf(String.valueOf(value));*/
+                //���������������������������������������������  start
+                Double sourceDataD = null;
+                if (value != null) {
+                     sourceDataD = Double.valueOf(String.valueOf(value));
+                }else{
+                    sourceDataD = 5.00d;
+                }
+                //���������������������������������������������  end
+                BigDecimal bg = new BigDecimal(sourceDataD);
+                bg = bg.setScale(2, BigDecimal.ROUND_FLOOR);
+                String sourceData = bg.toString();
                 //������������
                 //������������
                 if (!unitKey.equals(showUnitKey)) {//������������������������������������������������������������
@@ -70,22 +91,24 @@
                     //������������
                     String resultData = UnitConvertUtils.calculate(sourceData, formula);
                     if (resultData != null) {
-                        resultData+=showUnit;
-                    }else{//���������������������������null���������������������������������������������������������������
-                        resultData = sourceData+unit;
+                        resultData += showUnit;
+                    } else {//���������������������������null���������������������������������������������������������������
+                        resultData = sourceData + unit;
                     }
                     resultMessgae.put(sensor.getCode(), resultData);
                 } else {
                     //������������
-                    sourceData += showUnit;
+                    if(!showUnit.equals("���������"))
+                        sourceData = sourceData + " " + showUnit;
                     resultMessgae.put(sensor.getCode(), sourceData);
                 }
             }
-            resultMessgae.put("DataTime",message.get("DataTime"));
-            resultMessgae.put("time",message.get("time"));
-            resultMessgae.put("time1",message.get("time1"));
-            resultMessgae.put("serverTime",System.currentTimeMillis());
-            resultMessgae.put("serverStartTime",serverStartTime);
+            //���������������������������������
+            //resultMessgae.put("DataTime", message.get("DataTime"));
+            //resultMessgae.put("time", message.get("time"));
+            //resultMessgae.put("time1", message.get("time1"));
+            //resultMessgae.put("serverTime", System.currentTimeMillis());
+            //resultMessgae.put("serverStartTime", serverStartTime);
             socket.sendMessage(JSON.toJSONString(resultMessgae));
         }
     }

--
Gitblit v1.8.0