[update] 优化解析实时数据结构

signal-optimize-service/src/main/java/net/wanji/opt/kafka/ConsumerHandler.java

@@ -22,6 +22,8 @@ import org.springframework.util.CollectionUtils;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -48,13 +50,16 @@ public class ConsumerHandler implements KafkaListenerErrorHandler {
         try {
             String crossRealTimeLaneData = String.valueOf(record.value());
             if (StringUtils.isNotBlank(crossRealTimeLaneData)) {
-                List<CrossLaneDataRealTimePO> crossLaneDataRealTimePOS = JacksonUtils.getInstance().readValue(crossRealTimeLaneData, new TypeReference<List<CrossLaneDataRealTimePO>>() {});
-                if (!CollectionUtils.isEmpty(crossLaneDataRealTimePOS)) {
-                    crossLaneDataRealTimeMapper.deleteBatch(crossLaneDataRealTimePOS.stream().map(CrossLaneDataRealTimePO::getCrossId).collect(Collectors.toList()));
-                    crossLaneDataRealTimeMapper.insertBatch(crossLaneDataRealTimePOS);
-                    List<CrossLaneDataHistPO> crossLaneDataHistPOS = new ArrayList<>(crossLaneDataRealTimePOS.size());
-                    BeanListUtils.populateList(crossLaneDataRealTimePOS, crossLaneDataHistPOS, CrossLaneDataHistPO.class);
-                    crossLaneDataHistMapper.insertBatch(crossLaneDataHistPOS);
+                Map<String, List<CrossLaneDataRealTimePO>> map = JacksonUtils.getInstance().readValue(crossRealTimeLaneData, new TypeReference<Map<String, List<CrossLaneDataRealTimePO>>>() {});
+                if (Objects.nonNull(map) && !map.isEmpty()) {
+                    List<CrossLaneDataRealTimePO> crossLaneDataRealTimePOS = map.get("crossRealTimeLaneData");
+                    if (!CollectionUtils.isEmpty(crossLaneDataRealTimePOS)) {
+                        crossLaneDataRealTimeMapper.deleteBatch(crossLaneDataRealTimePOS.stream().map(CrossLaneDataRealTimePO::getCrossId).collect(Collectors.toList()));
+                        crossLaneDataRealTimeMapper.insertBatch(crossLaneDataRealTimePOS);
+                        List<CrossLaneDataHistPO> crossLaneDataHistPOS = new ArrayList<>(crossLaneDataRealTimePOS.size());
+                        BeanListUtils.populateList(crossLaneDataRealTimePOS, crossLaneDataHistPOS, CrossLaneDataHistPO.class);
+                        crossLaneDataHistMapper.insertBatch(crossLaneDataHistPOS);
+                    }
                 }
             }
             acknowledgment.acknowledge();
@@ -70,14 +75,18 @@ public class ConsumerHandler implements KafkaListenerErrorHandler {
         try {
             String crossRealTimeData = String.valueOf(record.value());
             if (StringUtils.isNotBlank(crossRealTimeData)) {
-                List<CrossDataRealtimePO> crossDataRealtimePOS = JacksonUtils.getInstance().readValue(crossRealTimeData, new TypeReference<List<CrossDataRealtimePO>>() {});
-                if (!CollectionUtils.isEmpty(crossDataRealtimePOS)) {
-                    crossDataRealTimeMapper.deleteBatch(crossDataRealtimePOS.stream().map(CrossDataRealtimePO::getCrossId).collect(Collectors.toList()));
-                    crossDataRealTimeMapper.insertBatch(crossDataRealtimePOS);
-                    List<CrossDataHistPO> crossDataHistPOS = new ArrayList<>(crossDataRealtimePOS.size());
-                    BeanListUtils.populateList(crossDataRealtimePOS, crossDataHistPOS, CrossDataHistPO.class);
-                    crossDataHistMapper.insertBatch(crossDataHistPOS);
+                Map<String, List<CrossDataRealtimePO>> map = JacksonUtils.getInstance().readValue(crossRealTimeData, new TypeReference<Map<String, List<CrossDataRealtimePO>>>() {});
+                if (Objects.nonNull(map) && !map.isEmpty()) {
+                    List<CrossDataRealtimePO> crossDataRealtimePOS = map.get("crossRealTimeData");
+                    if (!CollectionUtils.isEmpty(crossDataRealtimePOS)) {
+                        crossDataRealTimeMapper.deleteBatch(crossDataRealtimePOS.stream().map(CrossDataRealtimePO::getCrossId).collect(Collectors.toList()));
+                        crossDataRealTimeMapper.insertBatch(crossDataRealtimePOS);
+                        List<CrossDataHistPO> crossDataHistPOS = new ArrayList<>(crossDataRealtimePOS.size());
+                        BeanListUtils.populateList(crossDataRealtimePOS, crossDataHistPOS, CrossDataHistPO.class);
+                        crossDataHistMapper.insertBatch(crossDataHistPOS);
+                    }
                 }
+
             }
             acknowledgment.acknowledge();
         } catch (Exception e) {
@@ -92,13 +101,16 @@ public class ConsumerHandler implements KafkaListenerErrorHandler {
         try {
             String crossRealTimeDirData = String.valueOf(record.value());
             if (StringUtils.isNotBlank(crossRealTimeDirData)) {
-                List<CrossDirDataRealtimePO> crossDirDataRealtimePOS = JacksonUtils.getInstance().readValue(crossRealTimeDirData, new TypeReference<List<CrossDirDataRealtimePO>>() {});
-                if (!CollectionUtils.isEmpty(crossDirDataRealtimePOS)) {
-                    crossDirDataRealTimeMapper.deleteBatch(crossDirDataRealtimePOS.stream().map(CrossDirDataRealtimePO::getCrossId).collect(Collectors.toList()));
-                    crossDirDataRealTimeMapper.insertBatch(crossDirDataRealtimePOS);
-                    List<CrossDirDataHistPO> list = new ArrayList<>(crossDirDataRealtimePOS.size());
-                    BeanListUtils.populateList(crossDirDataRealtimePOS, list, CrossDirDataHistPO.class);
-                    crossDirDataHistMapper.insertBatch(list);
+                Map<String, List<CrossDirDataRealtimePO>> map = JacksonUtils.getInstance().readValue(crossRealTimeDirData, new TypeReference<Map<String, List<CrossDirDataRealtimePO>>>() {});
+                if (Objects.nonNull(map) && !map.isEmpty()) {
+                    List<CrossDirDataRealtimePO> crossDirDataRealtimePOS = map.get("crossRealTimeDirData");
+                    if (!CollectionUtils.isEmpty(crossDirDataRealtimePOS)) {
+                        crossDirDataRealTimeMapper.deleteBatch(crossDirDataRealtimePOS.stream().map(CrossDirDataRealtimePO::getCrossId).collect(Collectors.toList()));
+                        crossDirDataRealTimeMapper.insertBatch(crossDirDataRealtimePOS);
+                        List<CrossDirDataHistPO> list = new ArrayList<>(crossDirDataRealtimePOS.size());
+                        BeanListUtils.populateList(crossDirDataRealtimePOS, list, CrossDirDataHistPO.class);
+                        crossDirDataHistMapper.insertBatch(list);
+                    }
                 }
             }
             acknowledgment.acknowledge();
@@ -114,14 +126,17 @@ public class ConsumerHandler implements KafkaListenerErrorHandler {
         try {
             String crossRealTimeTurnData = String.valueOf(record.value());
             if (StringUtils.isNotBlank(crossRealTimeTurnData)) {
-                List<CrossTurnDataRealtimePO> crossTurnDataRealtimePOS = JacksonUtils.getInstance().readValue(crossRealTimeTurnData, new TypeReference<List<CrossTurnDataRealtimePO>>() {});
-                if (!CollectionUtils.isEmpty(crossTurnDataRealtimePOS)) {
-                    List<String> crossIds = crossTurnDataRealtimePOS.stream().map(CrossTurnDataRealtimePO::getCrossId).distinct().collect(Collectors.toList());
-                    crossTurnDataRealTimeMapper.deleteBatch(crossIds);
-                    crossTurnDataRealTimeMapper.insertBatch(crossTurnDataRealtimePOS);
-                    List<CrossTurnDataHistPO> list = new ArrayList<>(crossTurnDataRealtimePOS.size());
-                    BeanListUtils.populateList(crossTurnDataRealtimePOS, list, CrossTurnDataHistPO.class);
-                    crossTurnDataHistMapper.insertBatch(list);
+                Map<String, List<CrossTurnDataRealtimePO>> map = JacksonUtils.getInstance().readValue(crossRealTimeTurnData, new TypeReference<Map<String, List<CrossTurnDataRealtimePO>>>() {});
+                if (Objects.nonNull(map) && !map.isEmpty()) {
+                    List<CrossTurnDataRealtimePO> crossTurnDataRealtimePOS = map.get("crossRealTimeTurnData");
+                    if (!CollectionUtils.isEmpty(crossTurnDataRealtimePOS)) {
+                        List<String> crossIds = crossTurnDataRealtimePOS.stream().map(CrossTurnDataRealtimePO::getCrossId).distinct().collect(Collectors.toList());
+                        crossTurnDataRealTimeMapper.deleteBatch(crossIds);
+                        crossTurnDataRealTimeMapper.insertBatch(crossTurnDataRealtimePOS);
+                        List<CrossTurnDataHistPO> list = new ArrayList<>(crossTurnDataRealtimePOS.size());
+                        BeanListUtils.populateList(crossTurnDataRealtimePOS, list, CrossTurnDataHistPO.class);
+                        crossTurnDataHistMapper.insertBatch(list);
+                    }
                 }
             }
             acknowledgment.acknowledge();