ID:2023-01-03 MSG:行车情况提交最新代码,修改连接池大小

realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/dto/TravelInfo.java

@@ -31,6 +31,7 @@ public class TravelInfo implements Comparable<TravelInfo>  {
     private Long arrivedTime  ;//可选 到达时间，毫秒级别时间戳
     private Double arrivedSpeed  ;//可选 到达时速度，单位km/h
     private String inCrossLineId  ;//可选 进入路口时车道编号
+    private String flowDirection  ;//流向
     private Long inCrossTime  ;//可选 进入路口时间，毫秒级别时间戳
     private Double inSpeed  ;//可选 进入时速度，单位km/h
     private String outCrossLineId  ;//可选 通过路口时车道编号

realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelCarInfoCoProcessFunction.java

@@ -30,6 +30,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
     private static Logger logger = Logger.getLogger(TravelCarInfoCoProcessFunction.class);
 
+
     //定义状态，用于存储旅行信息，当周期信号数据过来的时候，触发计算
     //状态数据存储结构为   Map（路口id--》（Map（车道id-》（List（旅行信息））））） 一辆车只有一条旅行信息
     // 这样就把每个路口的数据进行分流，同时，对每个路口的数据，也按照了不同的车道进行了分流，同一个路口，同一个车道的数据存在一个list中
@@ -39,7 +40,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
     private  MapState<Integer, CycleSignalData> alreadyHandleCycleSignalDataState;  //已经处理过的周期数据，避免重复触发
 
-    private ListState<Long> cycleSectionState ;
+    private ListState<Long> cycleSectionState ; //周期数据切片 ，记录每个周期的开始时间，以便于进行对locations切分到不同周期，计算停车次数
     private static Map<String, Map<String, String>> dim_cnt_cross_lane_position;
 
     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@@ -62,6 +63,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
         ListStateDescriptor cycleSectionStateDescriptor = new ListStateDescriptor("cycleSectionState", TypeInformation.of(Long.class));
 
+        //待计算的周期数据
         StateTtlConfig stateTtlConfig4CycleData = StateTtlConfig
                 // 状态有效时间    3小时
                 .newBuilder(Time.hours(3))
@@ -73,6 +75,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                 .cleanupFullSnapshot()
                 .build();
         cycleSignalDataStateDescriptor.enableTimeToLive(stateTtlConfig4CycleData);
+        alreadyHandleCycleSignalDataStateDescriptor.enableTimeToLive(stateTtlConfig4CycleData);
 
         travelInfoState = getRuntimeContext().getMapState(travelInfoStateDescriptor);
         cycleSignalDataState = getRuntimeContext().getMapState(cycleSignalDataStateDescriptor);
@@ -94,9 +97,13 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     public void processElement1(TravelInfo value, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.Context ctx,
                                 Collector<List<TravelLineSinkInfo>> out) throws Exception {
 
+        if(value.getInCrossTime() < 0 || value.getArrivedTime() < 0){
+            logger.info("路口id"+value.getCrossId() +"车辆id"+value.getCarId() + "进入路口时间："+sdf.format(value.getInCrossTime())+"数据存在异常-作废");
+            return ;
+        }
         //  System.out.println("路口id"+value.getCrossId() +"车辆id"+value.getCarId() + "进入路口时间："+sdf.format(value.getInCrossTime()));
-        logger.info("路口id"+value.getCrossId() +"车辆id"+value.getCarId() + "进入路口时间："+sdf.format(value.getInCrossTime()));
-
+        logger.info("路口id"+value.getCrossId() +"车辆id"+value.getCarId() +
+                "进入路口时间："+sdf.format(value.getInCrossTime())+"到达时间"+sdf.format(value.getArrivedTime()));
         String crossId = value.getCrossId();
         Iterator<Map<String, List<TravelInfo>>> iterator = travelInfoState.values().iterator();
         //如果状态为空，则进行初始化
@@ -151,7 +158,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                                 Collector<List<TravelLineSinkInfo>> out)  {
 
         try {
-          //  System.out.println("路口id"+value.getCrossCode() +"周期id"+value.getCycleOrder() + "周期时间："+sdf.format(value.getBeginDateTime())+"~"+sdf.format(value.getEndDateTime()));
+            //System.out.println("路口id"+value.getCrossCode() +"周期id"+value.getCycleOrder() + "周期时间："+sdf.format(value.getBeginDateTime())+"~"+sdf.format(value.getEndDateTime()));
             logger.info("路口id"+value.getCrossCode() +"周期id"+value.getCycleOrder() + "周期时间："+sdf.format(value.getBeginDateTime())+"~"+sdf.format(value.getEndDateTime()));
 
             //判断当前周期是否执行过了,存在周期数据重复下发的情况，只触发一次计算
@@ -159,6 +166,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                 logger.info("周期数据重复----当前路口id是" + value.getCrossCode() + "当前周期已经计算完成了" + value.getCycleOrder());
                 return;
             }
+            //记录周期数据时间切片，以便划分locations
             cycleSectionState.add(value.getBeginDateTime());
 
             String crossCode = value.getCrossCode();
@@ -178,12 +186,17 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                     if (beforeCycleDataListMap != null) {
                         List<TravelLineSinkInfo> travelLineSinkInfos = handTwoStreamData(beforeCycleDataListMap, beforeCycle.getCycleOrder(),
                                 crossCode, beforeCycle.getBeginDateTime(), beforeCycle.getEndDateTime());
+
+                        // TODO 只有当前周期有数据sink，才算计算完成，否则继续等待触发
+                        if(travelLineSinkInfos.size()>0){
                             out.collect(travelLineSinkInfos);
                             // 当前周期已经处理完成了，kafka中再次来了该周期的数据不需要再次触发计算了
                             alreadyHandleCycleSignalDataState.put(beforeCycle.getCycleOrder(), beforeCycle);
                             listToRemove.add(beforeCycle.getCycleOrder());
                         }
                     }
+                }
+                // 清理待计算的周期数据，计算完成后进行移除
                 for (Integer i:listToRemove) {
                     cycleSignalDataState.remove(i);
                 }
@@ -194,10 +207,13 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             //如果，有旅行信息数据就计算，否则就不计算
             if (thisCycleDataListMap!=null && !thisCycleDataListMap.values().isEmpty()) {
                 List<TravelLineSinkInfo> travelLineSinkInfos = handTwoStreamData(thisCycleDataListMap, cycleOrder, crossCode, beginDateTime, endDateTime);
+                // TODO 只有当前周期有数据sink，才算计算完成，否则继续等待触发
+                if(travelLineSinkInfos.size()>0){
                     out.collect(travelLineSinkInfos);
                     // 当前周期已经处理完成了，，kafka中再次来了该周期的数据不需要再次触发计算了
                     alreadyHandleCycleSignalDataState.put(cycleOrder, value);
                     cycleSignalDataState.remove(cycleOrder);
+                }
             } else {
                 cycleSignalDataState.put(cycleOrder, value);
             }
@@ -361,14 +377,14 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                 List<TravelInfo> thisData = new ArrayList<>(); //属于当前周期的数据
                 for (TravelInfo t : allTravelInfos) {
                     //进入路口的时间属于这个周期
-                    if (t.getInCrossTime() >= beginDateTime && t.getInCrossTime() <= endDateTime) {
+                    if (t.getInCrossTime() >= beginDateTime && t.getInCrossTime() <= endDateTime && t.getLocations()!=null) {
                         //当前周期用到的数据
                         thisData.add(t);
                         //如果，进入路口时距小于周期开始时间，且已经超过30min，则清理这个数据
                     }
 
-                    else if(t.getInCrossTime()<beginDateTime && beginDateTime - t.getInCrossTime() > 1000*60*60 ) {
-                        logger.info("移除当前旅行信息-超过30分钟都没有触发计算"+t);
+                    else if(t.getInCrossTime()<beginDateTime && beginDateTime - t.getInCrossTime() > 1000*60*60*3 ) {
+                        logger.info("移除当前旅行信息-超过30分钟都没有触发计算"+"车辆id"+t.getCarId() + "进入路口时间："+sdf.format(t.getInCrossTime()));
                     }
                     else {
                         //当前周期没有用的的数据
@@ -411,7 +427,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
 
         //1、计算进入时间
-        long arriveTime = travel.getInCrossTime() - travel.getArrivedTime();
+        long passTime = travel.getInCrossTime() - travel.getArrivedTime();
 
         //2、获取方向
         String direction = dim_cnt_cross_lane_position.get(travel.getCrossId()).get(travel.getInCrossLineId()+"direction");
@@ -433,12 +449,13 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
         if (beforeCarInCrossTime == 0L) {
             carHeadTimeGap = 0L;
         } else {
-            carHeadTimeGap = travel.getInCrossTime() - beforeCarInCrossTime;
+//            取绝对值，避免出现插队导致车头时距变成负数
+            carHeadTimeGap = Math.abs(travel.getInCrossTime() - beforeCarInCrossTime);
         }
 
         TravelCarInfo travelCarInfo = new TravelCarInfo(travel.getCrossId(), cycleOrder,
                 travel.getCarId()
-                , arriveTime,travel.getInCrossTime(), travel.getInCrossLineId(), direction, flow_direction,averageSpeed,
+                , passTime,travel.getInCrossTime(), travel.getInCrossLineId(), direction, flow_direction,averageSpeed,
                 controlDelayTime, stopTimes, stopDelayTime, carHeadTimeGap);
 
         return travelCarInfo;
@@ -532,6 +549,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             list.add(l);
         }
 
+        // 这个会对周期切片的数据进行更新操作，最多保留最近3个周期的时间片
         if(list.size() > 3) {
             beginTimes = list.subList(list.size() - 3, list.size());
             cycleSectionState.update(beginTimes);
@@ -666,7 +684,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
         Iterator<Location> iterator = set.iterator();
         List<Location> speedList = new ArrayList<>();
         while (iterator.hasNext()) {
-            speedList.add(iterator.next());
+            Location next = iterator.next();
+            speedList.add(next);
         }
         Long stopDelayTime = 0L;
         Long stopStart = 0L;
@@ -677,15 +696,16 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                 stopStart = speedList.get(i).getDtTranjectory();
                 //从第二帧开始，只有前一帧速度大于10，且这一帧数据小于10的才算一次停车，连续小于3的视为一次停车
             } else {
-                //发生停车事件
-                if (speedList.get(i).getSpeed() < 3.0 && speedList.get(i - 1).getSpeed() >= 10.0) {
+                //发生停车事件 车速小于3.0视为停车
+                //   if (speedList.get(i).getSpeed() < 3.0 && speedList.get(i - 1).getSpeed() >= 10.0) {
+                if (speedList.get(i).getSpeed() < 3.0 && speedList.get(i - 1).getSpeed() >= 3.0) {
                     stopStart = speedList.get(i).getDtTranjectory();
                 }
                 //开始正常行驶
                 else {
                     if (stopStart != 0L && speedList.get(i).getSpeed() >= 3.0) {
                         stopDelayTime = (speedList.get(i).getDtTranjectory() - stopStart) + stopDelayTime;
-                        //开始正常行驶了，把这个停止起始时间置为0L
+
                         stopStart = 0L;
                     }
                     // 如果，是最后一帧数据并且是停止状态，且前面也是停止状态，则进行计算延迟时间
@@ -713,6 +733,21 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
         String inDirection = dim_cnt_cross_lane_position.get(travel.getCrossId()).get(inLineId+"direction");
         String outDirection = dim_cnt_cross_lane_position.get(travel.getCrossId()).get(outLineId+"direction");
 
+        if(inDirection==null || outDirection==null) {
+
+            String flowDirection = travel.getFlowDirection();
+            if(flowDirection==null){
+                return "未知";
+            } else {
+                String[] split = flowDirection.split("-");
+                inDirection=split[0];
+                outDirection=split[1];
+
+            }
+
+        }
+
+
         if(inDirection!=null&&inDirection.equals(outDirection)){
             return inDirection+"掉头";
         } else {
@@ -738,60 +773,5 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             }
             return "未知";
         }
-}
-
-
-
-
-    /**
-     * 将每个车道流向预聚合的数据存入mysql中去
-     * @param list
-     * @throws SQLException
-     *//*
-    private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
-
-
-            String sql = "INSERT INTO app_cross_line_travel_car_info" +
-                    "(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +
-                    " flow_direction, pass_numbers," +
-                    " last_car_inCross_time, average_pass_time, average_pass_speed, average_control_delay, " +
-                    "average_stop_delay, average_stop_times, average_car_head_time_gap" +
-                    ")" +
-                    "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?,?,?,?) ";
-            PreparedStatement preparedStatement = connection.prepareStatement(sql);
-
-            //TODO 这里的时间使用数据携带的时间，使用最后进入路口车辆的时间，这样在补数据时，可以补充到对应的分区下
-            Date date = new Date();
-            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-            SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd");
-            String statistic_time = sdf.format(date);
-
-
-            for (TravelLineSinkInfo t :list) {
-                // 使用事件数据的时间划分数据分区，这样可以进行补数操作
-                String record_date = sdf2.format(t.getLast_car_inCross_time());
-
-                preparedStatement.setString(1, record_date);
-                preparedStatement.setString(2, t.getCross_id());
-                preparedStatement.setString(3, statistic_time);
-                preparedStatement.setString(4, sdf.format(t.getCycle_begin_time()));
-                preparedStatement.setString(5, sdf.format(t.getCycle_end_time()));
-                preparedStatement.setString(6, t.getLane_id());
-                preparedStatement.setString(7, t.getCycle_id().toString());
-                preparedStatement.setString(8, t.getDirection());
-                preparedStatement.setString(9, t.getFlow_direction());
-                preparedStatement.setInt(10, t.getPass_numbers());
-                preparedStatement.setLong(11, t.getLast_car_inCross_time());
-                preparedStatement.setDouble(12, t.getAverage_pass_time());
-                preparedStatement.setDouble(13, t.getAverage_pass_speed());
-                preparedStatement.setDouble(14, t.getAverage_control_delay());
-                preparedStatement.setDouble(15, t.getAverage_stop_delay());
-                preparedStatement.setDouble(16, t.getAverage_stop_times());
-                preparedStatement.setDouble(17, t.getAverage_car_head_time_gap());
-                preparedStatement.addBatch();
-            }
-            preparedStatement.executeBatch();
-
-    }*/
-
+    }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelSituationAnalysisJob.java

@@ -14,6 +14,7 @@ import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.runtime.state.filesystem.FsStateBackend;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.ConnectedStreams;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
@@ -52,8 +53,12 @@ public class TravelSituationAnalysisJob {
 
             env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 
-            //设置重启策略
-            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
+            //这如果不指定使用FsStateBackend 则默认使用flink集群中的配置，rocksDB
+            env.setStateBackend(new FsStateBackend("file:///data/flink/state"));
+
+            //设置重启策略 这里不设置重启策略，由平台统一拉起任务
+            // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
+            env.setRestartStrategy(RestartStrategies.noRestart());
 
 
 

realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelLaneInfoSink.java

@@ -49,6 +49,8 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
     public void invoke(List<TravelLineSinkInfo> value, Context context) throws Exception {
         super.invoke(value, context);
 
+        //重新获取连接，避免超时的问题
+        connection = DruidConnectPoolUtils.getConnection();
         String sql = "INSERT INTO app_cross_line_travel_car_info" +
                 "(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +
                 " flow_direction, pass_numbers," +
@@ -92,5 +94,8 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
 
         logger.info("写入数据成功。。。。。"+ints.length+"...");
 
+        connection.close();
+
+
     }
 }

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DruidConnectPoolUtils.java

@@ -31,8 +31,8 @@ public class DruidConnectPoolUtils {
         dataSource.setUrl(dbHosts);
         dataSource.setUsername(username);
         dataSource.setPassword(password);
-        dataSource.setInitialSize(3);  //初始连接数，默认0
-        dataSource.setMaxActive(10);  //最大连接数，默认8
+        dataSource.setInitialSize(5);  //初始连接数，默认0
+        dataSource.setMaxActive(30);  //最大连接数，默认8
         dataSource.setMinIdle(5);  //最小闲置数
         dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
         dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除
@@ -47,8 +47,8 @@ public class DruidConnectPoolUtils {
         dataSource.setUrl(dbHosts);
         dataSource.setUsername(username);
         dataSource.setPassword(password);
-        dataSource.setInitialSize(3);  //初始连接数，默认0
-        dataSource.setMaxActive(10);  //最大连接数，默认8
+        dataSource.setInitialSize(5);  //初始连接数，默认0
+        dataSource.setMaxActive(30);  //最大连接数，默认8
         dataSource.setMinIdle(5);  //最小闲置数
         dataSource.setMaxWait(3000);//获取连接的最大等待时间，单位毫秒
         dataSource.setTestWhileIdle(true); //指明连接是否被空闲连接回收器(如果有)进行检验.如果检测失败,则连接将被从池中去除