Merge remote-tracking branch 'origin/master'

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

@@ -40,6 +40,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     private ValueState<Boolean> processCycle;
 
     private Connection connection;
+    //每个路口允许的最大时间数据
+    private static Long maxTravel=50000L;
 
     @Override
     public void open(Configuration parameters) throws Exception {
@@ -112,6 +114,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         List<TravelEvent> car = getCar(travelEvents,cycle);
         //根据车道分割过车数据
         Map<String, List<TravelEvent>> carForLane = getCarForLane(car);
+        //获取周期相位数据
         List<Stage> stageList = f1.getStageList();
         //获取相位关系
         for (Stage stage : stageList) {
@@ -124,6 +127,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         List<Map> positionList = dict.get("positionList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
         List<Map> positionNameList = dict.get("positionNameList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
         for (Map map : dictResult) {
+            //根据车道字典计算车道指标
             LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList);
             list.add(laneNorm);
             //交通流量为0得不参与计算
@@ -138,12 +142,16 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         if(laneNorms.size()>0){
             out.collect(laneNorms);
         }
+        //本次计算中未发生进入事件的数据
         List<TravelEvent> collect = car.stream().filter(recode -> StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList());
+        //计算中进入的数据
         List<TravelEvent> travelEvents1 = cycleTravelTemp.get(cycle.getCrossCode())==null?new ArrayList<>():cycleTravelTemp.get(cycle.getCrossCode());
-        //插入计算过程中缓存的数据
-        for (TravelEvent travelEvent:travelEvents1){
-            collect.add(travelEvent);
-        }
+        //未参与计算的事件数据
+        List<TravelEvent> futureEvents = travelEvents.stream().filter(a -> Long.parseLong(a.getEventTime()) > Long.parseLong(cycle.getEndDateTime())).collect(Collectors.toList());
+
+        //计算中进入的数据 未参与计算的事件数据 本次计算中未发生进入事件的数据进行合并 放入内存中 下周期继续计算
+        collect.addAll(travelEvents1);
+        collect.addAll(futureEvents);
         cycleTravel.put(f1.getCrossCode(),collect);
         cycleTravelTemp.put(f1.getCrossCode(),new ArrayList<>());
         processCycle.update(false);
@@ -160,13 +168,17 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         if (cycleTravelTemp.get(event.getCrossId()) == null) {
             cycleTravelTemp.put(event.getCrossId(), new ArrayList<>());
         }
-        //周期数据处理时，讲数据塞入临时内存中
+        //如果当前路口存留的事件数据积压 超过阈值 则进行清理操作 十分钟之前的数据丢弃
+        if(!processCycle.value() && cycleTravel.get(event.getCrossId()).size()>maxTravel){
+            List<TravelEvent> collect = cycleTravel.get(event.getCrossId()).stream().filter(a -> DateUtils.compareDataIsBefore10Min(a.getEventTime())).collect(Collectors.toList());
+            cycleTravelTemp.put(event.getCrossId(),collect);
+        }
+        //当周期数据处理时，将数据塞入临时内存中
         if (processCycle.value()) {
             cycleTravelTemp.get(event.getCrossId()).add(event);
         } else {
             cycleTravel.get(event.getCrossId()).add(event);
         }
-
     }
 
 
@@ -174,7 +186,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<LaneNorm>> out) throws Exception {
         super.onTimer(timestamp, ctx, out);
         System.out.println("触发定时器了。。。。");
-        //通过定时器清空字典表
+        //通过定时器清空字典表 防止字典表数据有更新
         dict.remove("phaseDirection");
         dict.remove("laneList");
         dict.remove("dictResult");
@@ -197,6 +209,21 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         return AllGreenTime + "";
     }
 
+    private static String gettEffectGreenTimeForDirction(List<Stage> stageList, String position,String direction) {
+        int AllGreenTime = 0;
+        for (Stage stage : stageList) {
+            List<StageForDirection> stageForDirectionList = stage.getStageForDirectionList();
+            for (StageForDirection dir : stageForDirectionList) {
+                if (position.equals(dir.getDirection()) && direction.equals(dir.getTraffic_flow_direction())) {
+                    AllGreenTime += stage.getValidGreen();
+                    //同一个相位累加一次即可
+                    break;
+                }
+            }
+        }
+        return AllGreenTime + "";
+    }
+
     /**
      * 提前计算所有相位允许的流向
      *
@@ -257,8 +284,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
                 //剩余承载力求和
                 double sum2 = collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum();
                 laneNorm.setResidualCapacity(new BigDecimal(sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
-                //有效绿灯时间 相位下此方向非右转得绿灯有效时间之和
-                laneNorm.setEffectGreenTime(gettEffectGreenTimeForPosition(cycle.getStageList(), position));
+                //有效绿灯时间 相位下此方向非右转的绿灯有效时间之和
+                laneNorm.setEffectGreenTime(gettEffectGreenTimeForDirction(cycle.getStageList(), position,name));
                 //识别空间空间占有率 车道长度需要交通流量为0得数据
                 int sum = list.stream().filter(recode -> position.equals(recode.getDirection()) && recode.getLaneName().indexOf(name) != -1).mapToInt(LaneNorm::getLaneLength).sum();
                 int sum1 = collect.stream().mapToInt(LaneNorm::getMaxQueueLength).sum();
@@ -485,7 +512,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
             if (laneCarMap.containsKey(laneId)) {
                 //获取这个车道所有的 过车数据
                 List<TravelEvent> travelEvents = laneCarMap.get(laneId);
-                //去周期内 每一秒的排队长度 取最大值
+                //周期内 每一秒的排队长度 取最大值
                 for(Long start = cycleStart; start <= cycleEnd; start += 1000) {
                     long start1 = start;
                     //通过每一秒分析出该车道停留的车数量
@@ -508,9 +535,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
 
     private static DruidPooledConnection getConnection() throws Exception {
         //修改为从连接池中取连接
-        DruidPooledConnection connection =  DruidConnectPoolUtils.getConnection();
+        //DruidPooledConnection connection =  DruidConnectPoolUtils.getConnection();
         //Connection connection = DorisUtils.getConnection();
-        return connection;
+        //return connection;
+        return null;
     }
 
     private static List<Map> getDictResult(Connection connection, String sql) throws Exception {
@@ -548,6 +576,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
             if (!DateUtils.compareDataIsBefore10Min(event.getEventTime())) {
                 continue;
             }
+            //事件数据发生在周期结束时间之后的 不参与计算
+            if (Long.parseLong(event.getEventTime())>Long.parseLong(cycle.getEndDateTime())) {
+                continue;
+            }
             if (!map.containsKey(event.getRecordId())) {
                 map.put(event.getRecordId(), event);
             }
@@ -575,6 +607,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         return lists;
     }
 
+    //将过车数据根据车道分组
     private static Map<String, List<TravelEvent>> getCarForLane(List<TravelEvent> cars) {
         Map<String, List<TravelEvent>> stringListHashMap = new HashMap<>();
         for (TravelEvent ev : cars) {

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

@@ -648,14 +648,12 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
      * @return
      */
     private Double getAverageSpeed(Map<String, TravelEvent> allEventOfCar) {
-
         //根据当前车辆的所有事件数据，计算平均速度
         Iterator<TravelEvent> iterator = allEventOfCar.values().iterator();
         double sumSpeed =0.0 ;
         while(iterator.hasNext()){
             sumSpeed= sumSpeed+iterator.next().getEventSpeed();
         }
-
         return sumSpeed/allEventOfCar.keySet().size() ;
     }
 
@@ -665,8 +663,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
      * @return
      */
     private Long getStopDelayTime(List<TravelEvent> travelEvents) {
-
-
         //按照事件发送的事件顺序进行排序后，再计算停车延误时间
         Set<TravelEvent> set = new TreeSet<>();
         for (TravelEvent t:travelEvents) {

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

@@ -11,6 +11,7 @@ import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.functions.FilterFunction;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.streaming.api.datastream.*;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -29,7 +30,9 @@ public class LaneNormJob {
     public static void main(String[] args) throws Exception {
         try {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-            env.setRestartStrategy(RestartStrategies.noRestart());
+            //env.setRestartStrategy(RestartStrategies.noRestart());
+            //设置重启策略
+            env.setRestartStrategy( RestartStrategies.fixedDelayRestart(10, Time.seconds(10)));
             DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob");
             DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob");
             KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {

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

@@ -365,6 +365,25 @@ public class DateUtils {
         return 0;
     }
 
+    public static int compare_dateTime(String DATE1, String DATE2) {
+
+
+        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
+        try {
+            Date dt1 = df.parse(DATE1);
+            Date dt2 = df.parse(DATE2);
+            if (dt1.getTime() > dt2.getTime()) {
+                return 1;
+            } else if (dt1.getTime() < dt2.getTime()) {
+                return -1;
+            } else {
+                return 0;
+            }
+        } catch (Exception exception) {
+            exception.printStackTrace();
+        }
+        return 0;
+    }
 
     public static String getBeforeTime(Calendar beforeTime,int before){
         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");