[add] 绿波城项目周期数据处理

wj-datacenter-service/src/main/java/net/wanji/datacenter/cache/CrossLaneInfoCache.java

@@ -8,6 +8,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -23,8 +24,10 @@ import java.util.stream.Collectors;
 public class CrossLaneInfoCache implements InitializingBean {
 
     public static final Map<String, LaneInfoPO> laneInfoMap = new HashMap<>();
-    /** key crossId11, value LaneInfoPO */
-    public static final Map<String, LaneInfoPO> crossIdLaneId2Map = new HashMap<>();
+    /** key crossId+dir */
+    public static final Map<String,LaneInfoPO> crossIdLaneId2Map = new HashMap<>();
+    /** key crossId, value dir */
+    public static final Map<String, List<Integer>> crossLaneDirMap = new HashMap<>();
 
     @Resource
     private LaneInfoMapper laneInfoMapper;
@@ -36,13 +39,21 @@ public class CrossLaneInfoCache implements InitializingBean {
             if (!CollectionUtils.isEmpty(laneInfoPOS)) {
                 Map<String, LaneInfoPO> crossMap = laneInfoPOS.stream().collect(Collectors.toMap(LaneInfoPO::getId, Function.identity(), (key1, key2) -> key2));
                 laneInfoMap.putAll(crossMap);
+                Map<Integer, List<LaneInfoPO>> dirLaneList = laneInfoPOS.stream().collect(Collectors.groupingBy(LaneInfoPO::getDir));
+                for (Map.Entry<Integer, List<LaneInfoPO>> entry : dirLaneList.entrySet()) {
+                    Integer dir = entry.getKey();
+                    List<LaneInfoPO> value = entry.getValue();
+                    for (LaneInfoPO laneInfoPO : value) {
+                        String laneInfoPOId = laneInfoPO.getId();
+                        String laneSortId = laneInfoPOId.substring(laneInfoPOId.length() - 2);
+                        crossIdLaneId2Map.put(crossId.concat(String.valueOf(dir)).concat(laneSortId), laneInfoPO);
+                    }
+                }
+                List<Integer> dirList = new ArrayList<>();
                 for (LaneInfoPO laneInfoPO : laneInfoPOS) {
-                    String currentCrossId = laneInfoPO.getCrossId();
-                    String laneId = laneInfoPO.getId();
-                    String laneId2 = laneId.substring(laneId.length() - 2);
-                    String dir = String.valueOf(laneInfoPO.getDir());
-                    crossIdLaneId2Map.put(currentCrossId.concat(dir).concat(laneId2), laneInfoPO);
+                    dirList.add(laneInfoPO.getDir());
                 }
+                crossLaneDirMap.put(crossId, dirList);
             }
         }
     }

wj-datacenter-service/src/main/java/net/wanji/datacenter/kafka/ConsumerHandler.java

@@ -33,33 +33,54 @@ public class ConsumerHandler implements KafkaListenerErrorHandler {
 
     @KafkaListener(topics = {"${kafka-consumer.lanePeriodicDataTopic}"}, groupId = "group")
     public void receiveLanePeriodicData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
-        //String originalData = String.valueOf(record.value());
-        String originalData = String.valueOf(laneStr);
+        String originalData = String.valueOf(record.value());
         try {
             dataProcessService.laneSave(originalData);
         } catch (Exception e) {
             log.error("车道周期实时数据转换异常", e);
             throw new Exception();
         }
-        // 修改逻辑，将保存逻辑改为将方向，转向数据返回kafka
-        //acknowledgment.acknowledge();
+        acknowledgment.acknowledge();
     }
 
     @KafkaListener(topics = {"${kafka-consumer.crossPeriodicDataTopic}"}, groupId = "group")
     public void receiveCrossPeriodicData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
-        //String originalData = String.valueOf(record.value());
-        String originalData = String.valueOf(crossStr);
+        String originalData = String.valueOf(record.value());
         try {
             dataProcessService.crossSave(originalData);
         } catch (Exception e) {
             log.error("车道周期实时数据转换异常", e);
             throw new Exception();
         }
-        // 修改逻辑，将保存逻辑改为将方向，转向数据返回kafka
-        //acknowledgment.acknowledge();
+        acknowledgment.acknowledge();
+    }
+
+    @KafkaListener(topics = {"${kafka-consumer.dirPeriodicDataTopic}"}, groupId = "group")
+    public void receiveDirPeriodicData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
+        String originalData = String.valueOf(record.value());
+        try {
+            dataProcessService.dirSave(originalData);
+        } catch (Exception e) {
+            log.error("车道周期实时数据转换异常", e);
+            throw new Exception();
+        }
+        acknowledgment.acknowledge();
+    }
+
+    @KafkaListener(topics = {"${kafka-consumer.turnPeriodicDataTopic}"}, groupId = "group")
+    public void receiveTurnPeriodicData(ConsumerRecord<Object, Object> record, Acknowledgment acknowledgment) throws Exception {
+        String originalData = String.valueOf(record.value());
+        try {
+            dataProcessService.turnSave(originalData);
+        } catch (Exception e) {
+            log.error("车道周期实时数据转换异常", e);
+            throw new Exception();
+        }
+        acknowledgment.acknowledge();
     }
 
 
+
     @Override
     @NonNull
     public Object handleError(Message<?> message, ListenerExecutionFailedException e) {
@@ -75,7 +96,4 @@ public class ConsumerHandler implements KafkaListenerErrorHandler {
         return KafkaListenerErrorHandler.super.handleError(message, exception, consumer);
     }
 
-    public static final String laneStr = "{\"orgCode\":\"370102\",\"timeStamp\":\"2024-11-15 15:40:00.000\",\"eventList\":[{\"crossId\":\"13MNM0B5OE0\",\"dir\":0,\"lane_num\":1,\"entry_type\":0,\"laneId\":\"1\",\"trafficFlow\":-1,\"trafficFlowA\":-1,\"trafficFlowB\":-1,\"trafficFlowC\":-1,\"trolleyEquivalent\":-1,\"laneFlowRate\":-1.0,\"meanV\":-1.0,\"vehicleNumsRatioMean\":-1.0,\"vehicleLengthRatioMean\":-1.0,\"timeOccupancy\":-1.0,\"staticQueueLengthMax\":-1.0,\"staticQueueLengthMin\":-1.0,\"dynamicQueueLengthMax\":-1.0,\"dynamicQueueLengthMin\":-1.0,\"lightGreenStartQueueLength\":-1.0,\"lightGreenFinishQueueLength\":-1.0,\"greenLightEfficiency\":-1.0,\"laneSaturationFlowRate\":-1,\"laneCapacity\":-1,\"laneSaturation\":-1.0,\"laneNoStopRate\":-1.0,\"laneOneStopRate\":-1.0,\"laneTwoStopRate\":-1.0,\"laneThreeStopRate\":-1.0,\"meanDelay\":-1.0,\"meanStopsNumber\":-1.0,\"meanLen\":0,\"timeHeadwaySectionMean\":-1.0,\"trafficInformation\":-1,\"overflowRate\":-1.0,\"nonMotorFlow\":-1.0,\"v85\":-1.0},{\"crossId\":\"13MNM0B5OR0\",\"dir\":2,\"lane_num\":2,\"entry_type\":-1,\"laneId\":\"2\",\"trafficFlow\":-1,\"trafficFlowA\":-1,\"trafficFlowB\":-1,\"trafficFlowC\":-1,\"trolleyEquivalent\":-1,\"laneFlowRate\":-1.0,\"meanV\":-1.0,\"vehicleNumsRatioMean\":-1.0,\"vehicleLengthRatioMean\":-1.0,\"timeOccupancy\":-1.0,\"staticQueueLengthMax\":-1.0,\"staticQueueLengthMin\":-1.0,\"dynamicQueueLengthMax\":-1.0,\"dynamicQueueLengthMin\":-1.0,\"lightGreenStartQueueLength\":-1.0,\"lightGreenFinishQueueLength\":-1.0,\"greenLightEfficiency\":-1.0,\"laneSaturationFlowRate\":-1,\"laneCapacity\":-1,\"laneSaturation\":-1.0,\"laneNoStopRate\":-1.0,\"laneOneStopRate\":-1.0,\"laneTwoStopRate\":-1.0,\"laneThreeStopRate\":-1.0,\"meanDelay\":-1.0,\"meanStopsNumber\":-1.0,\"meanLen\":-1.0,\"timeHeadwaySectionMean\":-1.0,\"trafficInformation\":-1,\"overflowRate\":-1.0,\"nonMotorFlow\":-1.0,\"v85\":-1.0}]}";
-
-    public static final String crossStr = "{\"orgCode\":\"370112\",\"timeStamp\":\"2024-11-15 20:10:00.000\",\"eventList\":[{\"crossId\":\"13N7H0B63Q0\",\"status\":-1,\"type\":-1,\"trafficIndex\":-1.0,\"startTime\":\"2024-11-15 20:05:00\",\"duration\":-1,\"isUnbalance\":0,\"isSpillover\":-1,\"emptyPass\":-1,\"isCongestion\":-1,\"unbalanceIndex\":null,\"spilloverIndex\":-1.0,\"congestionIndex\":-1.0,\"unbalanceDirs\":\"-1\",\"spilloverDirs\":\"-1\",\"emptyDirTurn\":\"-1\",\"congestionDirs\":\"-1\",\"flow\":-1,\"flowRate\":-1.0,\"speed\":-1.0,\"queueLength\":-1.0,\"stopTimes\":-1.0,\"delayTime\":-1,\"sturation\":-1.0,\"serviceLevel\":\"Z\",\"trafficState\":-1,\"strategy\":-1,\"strategyDuration\":-1,\"optimizeCount\":-1,\"optimizeSeconds\":-1,\"batchTime\":1731672601},{\"crossId\":\"13N920B63K0\",\"status\":-1,\"type\":-1,\"trafficIndex\":-1.0,\"startTime\":\"2024-11-15 20:05:00\",\"duration\":-1,\"isUnbalance\":0,\"isSpillover\":-1,\"emptyPass\":-1,\"isCongestion\":-1,\"unbalanceIndex\":null,\"spilloverIndex\":-1.0,\"congestionIndex\":-1.0,\"unbalanceDirs\":\"-1\",\"spilloverDirs\":\"-1\",\"emptyDirTurn\":\"-1\",\"congestionDirs\":\"-1\",\"flow\":-1,\"flowRate\":-1.0,\"speed\":-1.0,\"queueLength\":-1.0,\"stopTimes\":-1.0,\"delayTime\":-1,\"sturation\":-1.0,\"serviceLevel\":\"Z\",\"trafficState\":-1,\"strategy\":-1,\"strategyDuration\":-1,\"optimizeCount\":-1,\"optimizeSeconds\":-1,\"batchTime\":1731672601}]}";
 }

wj-datacenter-service/src/main/java/net/wanji/datacenter/pojo/dto/CrossPeriodDirDTO.java

+package net.wanji.datacenter.pojo.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+import net.wanji.databus.po.CrossDirDataRealtimePO;
+
+import java.util.List;
+
+/**
+ * @author duanruiming
+ * @date 2024/11/20 15:49
+ */
+@Data
+public class CrossPeriodDirDTO {
+
+    @JsonProperty("eventList")
+    private List<CrossDirDataRealtimePO> eventList;
+    /**
+     * 车道数量
+     */
+    private Integer laneNum;
+    /**
+     * 全域编号
+     */
+    private String orgCode;
+    /**
+     * 数据生成时间: yyyy-MM-dd HH:mm:ss:SSS
+     */
+    private String timeStamp;
+}

wj-datacenter-service/src/main/java/net/wanji/datacenter/pojo/dto/LanePeriodicDataDTO.java → wj-datacenter-service/src/main/java/net/wanji/datacenter/pojo/dto/CrossPeriodLaneDTO.java

@@ -13,7 +13,7 @@ import java.util.List;
  * @date 2023/03/09 17:02
  */
 @Data
-public class LanePeriodicDataDTO {
+public class CrossPeriodLaneDTO {
 
     /**
      * 车道事件集合

wj-datacenter-service/src/main/java/net/wanji/datacenter/pojo/dto/CrossPeriodTurnDTO.java

+package net.wanji.datacenter.pojo.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+import net.wanji.databus.po.CrossTurnDataRealtimePO;
+
+import java.util.List;
+
+/**
+ * @author duanruiming
+ * @date 2024/11/20 15:52
+ */
+@Data
+public class CrossPeriodTurnDTO {
+    @JsonProperty("eventList")
+    private List<CrossTurnDataRealtimePO> eventList;
+    /**
+     * 车道数量
+     */
+    private Integer laneNum;
+    /**
+     * 全域编号
+     */
+    private String orgCode;
+    /**
+     * 数据生成时间: yyyy-MM-dd HH:mm:ss:SSS
+     */
+    private String timeStamp;
+}

wj-datacenter-service/src/main/java/net/wanji/datacenter/service/DataProcessService.java

@@ -20,4 +20,22 @@ public interface DataProcessService {
      */
     void crossSave(String originalData) throws Exception;
 
+
+    /**
+     * 方向信息处理存库
+     * @param originalData
+     * @throws Exception
+     */
+    void dirSave(String originalData) throws Exception;
+
+
+    /**
+     * 转向信息处理存库
+     * @param originalData
+     * @throws Exception
+     */
+    void turnSave(String originalData) throws Exception;
+
+
+
 }

wj-datacenter-service/src/main/java/net/wanji/datacenter/service/impl/DataProcessServiceImpl.java

@@ -2,14 +2,14 @@ package net.wanji.datacenter.service.impl;
 
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
-import net.wanji.databus.dao.mapper.CrossDataRealtimeMapper;
-import net.wanji.databus.dao.mapper.CrossLaneDataRealTimeMapper;
-import net.wanji.databus.po.CrossDataRealtimePO;
-import net.wanji.databus.po.CrossLaneDataRealTimePO;
-import net.wanji.databus.po.LaneInfoPO;
+import net.wanji.common.utils.tool.BeanListUtils;
+import net.wanji.databus.dao.mapper.*;
+import net.wanji.databus.po.*;
 import net.wanji.datacenter.cache.CrossLaneInfoCache;
 import net.wanji.datacenter.pojo.dto.CrossPeriodDataDTO;
-import net.wanji.datacenter.pojo.dto.LanePeriodicDataDTO;
+import net.wanji.datacenter.pojo.dto.CrossPeriodDirDTO;
+import net.wanji.datacenter.pojo.dto.CrossPeriodLaneDTO;
+import net.wanji.datacenter.pojo.dto.CrossPeriodTurnDTO;
 import net.wanji.datacenter.service.DataProcessService;
 import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Service;
@@ -20,6 +20,7 @@ import javax.annotation.Resource;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.*;
+import java.util.stream.Collectors;
 
 /**
  * @author duanruiming
@@ -32,7 +33,20 @@ public class DataProcessServiceImpl implements DataProcessService {
     @Resource
     private CrossLaneDataRealTimeMapper crossLaneDataRealTimeMapper;
     @Resource
+    private CrossLaneDataHistMapper crossLaneDataHistMapper;
+    @Resource
     private CrossDataRealtimeMapper crossDataRealtimeMapper;
+    @Resource
+    private CrossDataHistMapper crossDataHistMapper;
+    @Resource
+    private CrossDirDataRealtimeMapper crossDirDataRealTimeMapper;
+    @Resource
+    private CrossDirDataHistMapper crossDirDataHistMapper;
+    @Resource
+    private CrossTurnDataRealtimeMapper crossTurnDataRealTimeMapper;
+    @Resource
+    private CrossTurnDataHistMapper crossTurnDataHistMapper;
+
 
     /**
      * 车道信息处理存库
@@ -41,41 +55,61 @@ public class DataProcessServiceImpl implements DataProcessService {
     @Override
     public void laneSave(String originalData) throws Exception {
         try {
-            Map<String, LaneInfoPO> laneInfoMap = CrossLaneInfoCache.crossIdLaneId2Map;
-            LanePeriodicDataDTO LanePeriodicDataDTO = JSONObject.parseObject(originalData, LanePeriodicDataDTO.class);
-            String timeStamp = LanePeriodicDataDTO.getTimeStamp();
+            CrossPeriodLaneDTO CrossPeriodLaneDTO = JSONObject.parseObject(originalData, CrossPeriodLaneDTO.class);
+            String timeStamp = CrossPeriodLaneDTO.getTimeStamp();
             SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
-            List<LanePeriodicDataDTO.CrossLaneDataRealTimeDTO> eventList = LanePeriodicDataDTO.getEventList();
-            List<CrossLaneDataRealTimePO> insertList = getInsertLaneList(laneInfoMap, timeStamp, dateFormat, eventList);
-            //crossLaneDataRealTimeMapper.insertBatch(insertList);
+            List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO> eventList = CrossPeriodLaneDTO.getEventList();
+            List<CrossLaneDataRealTimePO> insertList = getInsertLaneList(timeStamp, dateFormat, eventList);
+            // 数据库操作
+            if (!CollectionUtils.isEmpty(insertList)) {
+                crossLaneDataRealTimeMapper.deleteBatch(insertList.stream().map(CrossLaneDataRealTimePO::getCrossId).collect(Collectors.toList()));
+                crossLaneDataRealTimeMapper.insertBatch(insertList);
+                List<CrossLaneDataHistPO> crossLaneDataHistPOS = new ArrayList<>(insertList.size());
+                BeanListUtils.populateList(insertList, crossLaneDataHistPOS, CrossLaneDataHistPO.class);
+                crossLaneDataHistMapper.insertBatch(crossLaneDataHistPOS);
+            }
         } catch (Exception e) {
             log.error("车道数据处理入库异常", e);
             throw new Exception(e);
         }
     }
 
-    private static List<CrossLaneDataRealTimePO> getInsertLaneList(Map<String, LaneInfoPO> laneInfoMap, String timeStamp, SimpleDateFormat dateFormat, List<LanePeriodicDataDTO.CrossLaneDataRealTimeDTO> eventList) throws ParseException {
+    private static List<CrossLaneDataRealTimePO> getInsertLaneList(String timeStamp, SimpleDateFormat dateFormat, List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO> eventList) throws ParseException {
         List<CrossLaneDataRealTimePO> insertList = new ArrayList<>();
         if (!CollectionUtils.isEmpty(eventList)) {
-            for (LanePeriodicDataDTO.CrossLaneDataRealTimeDTO crossLaneDataRealTimeDTO : eventList) {
-                String crossId = crossLaneDataRealTimeDTO.getCrossId();
-                Integer dir = crossLaneDataRealTimeDTO.getDir();
-                Integer laneNum = crossLaneDataRealTimeDTO.getLane_num();
-                // 方向=神思dir + 1；车道号=神思laneNum + 10
-                String key = crossId.concat(String.valueOf(dir + 1)).concat(String.valueOf(laneNum + 10));
-                LaneInfoPO laneInfoPO = laneInfoMap.get(key);
-                if (Objects.isNull(laneInfoPO)) {
-                    continue;
+            Map<String, LaneInfoPO> crossIdLaneId2Map = CrossLaneInfoCache.crossIdLaneId2Map;
+            Map<String, List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO>> crossLaneMap = eventList.stream().collect(Collectors.groupingBy(CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO::getCrossId));
+            // 路口级别
+            for (Map.Entry<String, List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO>> entry : crossLaneMap.entrySet()) {
+                String crossId = entry.getKey();
+                List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO> eventListValue = entry.getValue();
+                Map<Integer, List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO>> dirLaneMap = eventListValue.stream().collect(Collectors.groupingBy(CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO::getDir));
+                // 方向级别
+                for (Map.Entry<Integer, List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO>> dirLaneEntry : dirLaneMap.entrySet()) {
+                    Integer dir = dirLaneEntry.getKey();
+                    List<CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO> value = dirLaneEntry.getValue();
+                    // 当前方向车道数量
+                    int size = value.size();
+                    // 车道级别
+                    for (CrossPeriodLaneDTO.CrossLaneDataRealTimeDTO crossLaneDataRealTimeDTO : value) {
+                        Integer wjLaneId = size - dir + 1 + 10;
+                        // 神思车道号与万集车道号镜像
+                        String key = crossId.concat(String.valueOf(dir + 1)).concat(String.valueOf(wjLaneId));
+                        LaneInfoPO laneInfoPO = crossIdLaneId2Map.get(key);
+                        if (Objects.isNull(laneInfoPO)) {
+                            continue;
+                        }
+                        String laneId = laneInfoPO.getId();
+                        crossLaneDataRealTimeDTO.setId(laneId);
+                        CrossLaneDataRealTimePO crossLaneDataRealTime = new CrossLaneDataRealTimePO();
+                        BeanUtils.copyProperties(crossLaneDataRealTimeDTO, crossLaneDataRealTime);
+                        Date startDate = dateFormat.parse(timeStamp);
+                        crossLaneDataRealTime.setStartTime(startDate);
+                        Integer batchTime = Integer.valueOf(String.valueOf(startDate.getTime()).substring(0, 10));
+                        crossLaneDataRealTime.setBatchTime(batchTime);
+                        insertList.add(crossLaneDataRealTime);
+                    }
                 }
-                String laneId = laneInfoPO.getId();
-                crossLaneDataRealTimeDTO.setId(laneId);
-                CrossLaneDataRealTimePO crossLaneDataRealTime = new CrossLaneDataRealTimePO();
-                BeanUtils.copyProperties(crossLaneDataRealTimeDTO, crossLaneDataRealTime);
-                Date startDate = dateFormat.parse(timeStamp);
-                crossLaneDataRealTime.setStartTime(startDate);
-                Integer batchTime = Integer.valueOf(String.valueOf(startDate.getTime()).substring(0, 10));
-                crossLaneDataRealTime.setBatchTime(batchTime);
-                insertList.add(crossLaneDataRealTime);
             }
         }
         return insertList;
@@ -89,7 +123,60 @@ public class DataProcessServiceImpl implements DataProcessService {
             String timeStamp = crossPeriodDataDTO.getTimeStamp();
             SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
             List<CrossDataRealtimePO> eventList = crossPeriodDataDTO.getEventList();
-            //crossDataRealtimeMapper.insertBatch(eventList);
+            if (!CollectionUtils.isEmpty(eventList)) {
+                List<CrossDataRealtimePO> crossDataRealtimePOS = eventList;
+                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);
+            }
+        } catch (Exception e) {
+            log.error("路口实时数据存库处理异常：{}", e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Transactional(rollbackFor = Exception.class)
+    @Override
+    public void dirSave(String originalData) throws Exception {
+        try {
+            CrossPeriodDirDTO crossPeriodDirDTO = JSONObject.parseObject(originalData, CrossPeriodDirDTO.class);
+            String timeStamp = crossPeriodDirDTO.getTimeStamp();
+            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
+            List<CrossDirDataRealtimePO> eventList = crossPeriodDirDTO.getEventList();
+            if (!CollectionUtils.isEmpty(eventList)) {
+                List<CrossDirDataRealtimePO> crossDirDataRealtimePOS = eventList;
+                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);
+            }
+        } catch (Exception e) {
+            log.error("路口实时数据存库处理异常：{}", e);
+            throw new RuntimeException(e);
+        }
+    }
+
+
+    @Transactional(rollbackFor = Exception.class)
+    @Override
+    public void turnSave(String originalData) throws Exception {
+        try {
+            CrossPeriodTurnDTO crossPeriodTurnDTO = JSONObject.parseObject(originalData, CrossPeriodTurnDTO.class);
+            String timeStamp = crossPeriodTurnDTO.getTimeStamp();
+            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
+            List<CrossTurnDataRealtimePO> eventList = crossPeriodTurnDTO.getEventList();
+            if (!CollectionUtils.isEmpty(eventList)) {
+                List<CrossTurnDataRealtimePO> crossTurnDataRealtimePOS = eventList;
+                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);
+            }
         } catch (Exception e) {
             log.error("路口实时数据存库处理异常：{}", e);
             throw new RuntimeException(e);