[update] 添加异常处理

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

@@ -45,69 +45,90 @@ public class ConsumerHandler implements KafkaListenerErrorHandler {
     @Transactional
     @KafkaListener(topics = {"cross_real_time_lane_data"})
     public void receiveCrossRealTimeLaneData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
-        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);
+        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);
+                }
             }
+            acknowledgment.acknowledge();
+        } catch (Exception e) {
+            log.error("kafka消费数据插入路口车道实时表异常", e);
+            throw new Exception(e);
         }
-        acknowledgment.acknowledge();
     }
 
     @Transactional
     @KafkaListener(topics = {"cross_real_time_data"})
     public void receiveCrossRealTimeData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
-        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);
+        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);
+                }
             }
+            acknowledgment.acknowledge();
+        } catch (Exception e) {
+            log.error("kafka消费数据插入路口实时表异常", e);
+            throw new Exception(e);
         }
-        acknowledgment.acknowledge();
     }
 
     @Transactional
     @KafkaListener(topics = {"cross_real_time_dir_data"})
     public void receiveCrossRealTimeDirData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
-        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);
+        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);
+                }
             }
+            acknowledgment.acknowledge();
+        } catch (Exception e) {
+            log.error("kafka消费数据插入路口方向实时表异常", e);
+            throw new Exception(e);
         }
-        acknowledgment.acknowledge();
     }
 
     @Transactional
     @KafkaListener(topics = {"cross_real_time_turn_data"})
     public void receiveCrossRealTimeTurnData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
-        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)) {
-                crossTurnDataRealTimeMapper.deleteBatch(crossTurnDataRealtimePOS.stream().map(CrossTurnDataRealtimePO::getCrossId).collect(Collectors.toList()));
-                crossTurnDataRealTimeMapper.insertBatch(crossTurnDataRealtimePOS);
-                List<CrossTurnDataHistPO> list = new ArrayList<>(crossTurnDataRealtimePOS.size());
-                BeanListUtils.populateList(crossTurnDataRealtimePOS, list, CrossTurnDataHistPO.class);
-                crossTurnDataHistMapper.insertBatch(list);
+        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);
+                }
             }
+            acknowledgment.acknowledge();
+        } catch (Exception e) {
+            log.error("kafka消费数据插入路口转向实时表异常", e);
+            throw new Exception(e);
         }
-        acknowledgment.acknowledge();
     }
 
     @Override