车道指标相关计算逻辑

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

@@ -38,7 +38,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     //存储mysql维表数据
     private MapState<String, List<Map>> dict;
     //
-    private ValueState<Boolean> processCycle;
+    private MapState<String, Boolean> processCycle;
 
     Connection connection;
     //每个路口允许的最大时间数据
@@ -54,9 +54,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         cycleTravel = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravel", String.class, Object.class));
         cycleTravelTemp = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravelTemp", String.class, Object.class));
         dict = getRuntimeContext().getMapState(new MapStateDescriptor("dict", String.class, Object.class));
-        processCycle = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("processCycle", Boolean.class));
-        connection=DorisUtils.getConnection();
-
+        processCycle = getRuntimeContext().getMapState(new MapStateDescriptor("processCycle",String.class, Object.class));
+        //connection=DorisUtils.getConnection();
         StateTtlConfig ttlConfig = StateTtlConfig
                 .newBuilder(Time.hours(3))
                 .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
@@ -95,7 +94,11 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
 
     @Override
     public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
-        processCycle.update(true);
+        //状态位修改为一个路口一个状态位
+        if(!processCycle.contains(cycle.getCrossCode())){
+            processCycle.put(cycle.getCrossCode(),false);
+        }
+        processCycle.put(cycle.getCrossCode(),true);
         List<LaneNorm> laneNorms = new ArrayList<>();
         if(!dayCycle.contains(DateUtils.toDateStr(new Date())+"-"+cycle.getCrossCode()+"-"+cycle.getCycleOrder())){
             dayCycle.put(DateUtils.toDateStr(new Date())+"-"+cycle.getCrossCode()+"-"+cycle.getCycleOrder(),"true");
@@ -106,11 +109,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         //初始化字典数据
         if (dict.get("phaseDirection") == null || dict.get("laneList") == null ||  dict.get("dictResult") == null ||  dict.get("positionList") == null ||  dict.get("positionNameList") == null) {
             loggerLevel = applicationUtils.getLoggerLevel();
-            if(!DataSourceUtils.isDbConnected(connection)){
-                connection=DorisUtils.getConnection();
-            }
+            connection=DorisUtils.getConnection();
+            Statement statement = connection.createStatement();
             //查询相位方向信息
-            dict.put("phaseDirection", getDictResult(connection, "SELECT\n" +
+            dict.put("phaseDirection", getDictResult(statement, "SELECT\n" +
                     "  a.ID id,\n" +
                     "  a.Direction direction,\n" +
                     "  b.Name traffic_flow_direction,\n" +
@@ -122,26 +124,31 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
                     "LEFT JOIN dim_dic_phase c\n" +
                     "  on a.PhaseId = c.pid"));
             //车道道路流向关系
-            dict.put("laneList", getDictResult(connection, "SELECT cross_id,lane_id,turn_direction as name,position,lane_length FROM dim_cnt_cross_lane_position\n" +
+            dict.put("laneList", getDictResult(statement, "SELECT cross_id,lane_id,turn_direction as name,position,lane_length FROM dim_cnt_cross_lane_position\n" +
                     "                where out_in_type='in'"));
             //获取所有车道信息
-            dict.put("dictResult", getDictResult(connection, "select a.*,b.saturation_flow_rate from (\n" +
+            dict.put("dictResult", getDictResult(statement, "select a.*,b.saturation_flow_rate from (\n" +
                     "                    SELECT cross_id,lane_id,position,lane_length,GROUP_CONCAT(turn_direction) as name FROM dim_cnt_cross_lane_position\n" +
                     "                                        where out_in_type='in' group By lane_id,cross_id,position,lane_length,cross_id order by cross_id,lane_id)a left join app_atomic_lane_saturation_flow_rate b\n" +
                     "                                        on a.lane_id=b.lane_id and a.cross_id=b.cross_id\t\t" +
                     "                                        where 1=1\n" +
                     "                                        and  b.inout_name='in' order by b.lane_id asc"));
             //查询路口所有的方向
-            dict.put("positionList", getDictResult(connection, "SELECT cross_id,position FROM dim_cnt_cross_lane_position\n" +
+            dict.put("positionList", getDictResult(statement, "SELECT cross_id,position FROM dim_cnt_cross_lane_position\n" +
                     "                                    where out_in_type='in' GROUP BY position,cross_id"));
             //查询路口所有的流向
-            dict.put("positionNameList", getDictResult(connection, " SELECT position,turn_direction as name,cross_id  FROM dim_cnt_cross_lane_position\n" +
+            dict.put("positionNameList", getDictResult(statement, " SELECT position,turn_direction as name,cross_id  FROM dim_cnt_cross_lane_position\n" +
                     "                            where out_in_type='in' GROUP BY position,turn_direction,cross_id"));
 
             long currentProcessingTime = ctx.timerService().currentProcessingTime();
             //12小时触发一次
             ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+12*60*60*1000);
             //ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+1*60*1000);
+            //查完字典表后 直接释放连接，下一次再获取即可
+            if(connection!=null){
+                statement.close();
+                connection.close();
+            }
         }
         Cycle f1 = cycle;
         // 代表周期已到，需要进行相应的指标计算了
@@ -178,6 +185,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         getPositionNorm(list, positionList, f1, laneNorms);
         //获取所有流向的指标
         getPositionNameNorm(list, positionNameList, f1, laneNorms,loggerLevel);
+
+        String afterTimeSecond = DateUtils.getAfterTimeSecond(20);
         if(laneNorms.size()>0){
             out.collect(laneNorms);
         }
@@ -192,14 +201,16 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         collect.addAll(futureEvents);
         cycleTravel.put(f1.getCrossCode(),collect);
         cycleTravelTemp.put(f1.getCrossCode(),new ArrayList<>());
-        processCycle.update(false);
+        processCycle.put(cycle.getCrossCode(),false);
     }
 
     @Override
     public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
-        if(processCycle.value()==null){
-            processCycle.update(false);
+        if(!processCycle.contains(event.getCrossId())){
+            processCycle.put(event.getCrossId(),false);
         }
+        boolean b = processCycle.get(event.getCrossId());
+        //System.out.println("路口->"+event.getCrossId()+" 处理状态->"+b);
         if (cycleTravel.get(event.getCrossId()) == null) {
             cycleTravel.put(event.getCrossId(), new ArrayList<>());
         }
@@ -207,12 +218,13 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
             cycleTravelTemp.put(event.getCrossId(), new ArrayList<>());
         }
         //如果当前路口存留的事件数据积压 超过阈值 则进行清理操作 十分钟之前的数据丢弃
-        if(!processCycle.value() && cycleTravel.get(event.getCrossId()).size()>maxTravel){
+        if(!b && 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()) {
+        if (b) {
             cycleTravelTemp.get(event.getCrossId()).add(event);
         } else {
             cycleTravel.get(event.getCrossId()).add(event);
@@ -399,28 +411,6 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     public static void main(String[] args) {
         try {
             Connection connection = getConnection();
-            boolean dbConnected = DataSourceUtils.isDbConnected(connection);
-
-            List<Map> dictResult = getDictResult(connection, "SELECT\n" +
-                    "  a.ID id,\n" +
-                    "  a.Direction direction,\n" +
-                    "  b.Name traffic_flow_direction,\n" +
-                    "  c.PID phase_no,\n" +
-                    "  c.PName phase_name\n" +
-                    "FROM dim_phase_movement a\n" +
-                    "LEFT JOIN dim_traffic_flow_direction b\n" +
-                    "  on a.TrafficFlowId = b.ID\n" +
-                    "LEFT JOIN dim_dic_phase c\n" +
-                    "  on a.PhaseId = c.ID");
-            connection.close();
-            Cycle cycle = setCycle();
-            Cycle cycle1 = cycle;
-            for (Stage stage : cycle1.getStageList()) {
-                stage.setStageForDirectionList(getStageForDiretion(stage, dictResult));
-            }
-            String s = gettEffectGreenTimeForDirction(cycle, "北", "直行","debug");
-            //System.out.println(s);
-
         }catch (Exception e){
             System.out.println(e);
         }
@@ -700,8 +690,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         return connection;
     }
 
-    private static List<Map> getDictResult(Connection connection, String sql) throws Exception {
-        Statement statement = connection.createStatement();
+    private static List<Map> getDictResult(Statement statement, String sql) throws Exception {
+
         ResultSet resultset = statement.executeQuery(sql);
         List list = resultSetToList(resultset);
         return list;

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

@@ -35,6 +35,9 @@ public class LaneNormJob {
             //env.setRestartStrategy( RestartStrategies.fixedDelayRestart(10, Time.seconds(10)));
             DataStream<String> cycleStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"signal_cycle_data","laneNormJob1");
             DataStream<String> carStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"trips_event_info","laneNormJob1");
+            //测试用topic
+            //DataStream<String> cycleStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"s_data2","laneNormJob1");
+            //DataStream<String> carStringStream = FlinkUtils.createKafkaStreamHuangzhun(args, env,"t_info2","laneNormJob1");
             KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
                 @Override
                 public String getKey(Cycle cycle) throws Exception {
@@ -79,12 +82,12 @@ public class LaneNormJob {
             //根据corssid 分组 保证同一路口的数据进行一个进程
             DataStream<Cycle> cycleDataStreamSource = cycleStringStream.map(getCycleFunction).assignTimestampsAndWatermarks(cycleWatermarkStrategy).keyBy(cycleKeySelect);
             DataStream<TravelEvent> carStream = carStringStream.map(getTravelEventFunction).assignTimestampsAndWatermarks(travelEventWatermarkStrategy).keyBy(carKeySelector).filter(filterFunction);
-
             ConnectedStreams<Cycle, TravelEvent> cycleTravelEventConnectedStreams = cycleDataStreamSource.connect(carStream).keyBy(cycle -> cycle.getCrossCode(), travelEvent -> travelEvent.getCrossId());
             SingleOutputStreamOperator<List<LaneNorm>> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4);
             process.addSink(new LaneNormSink()).name(" ");
             env.execute("laneNormJob");
         }catch (Exception e){
+            e.printStackTrace();
             System.out.println("程序异常关闭->"+e);
         }
 

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

@@ -52,12 +52,6 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
         if(laneNorms.size()==0){
             return;
         }
-        /**
-         * 添加连接判断 防止jdbc 失效
-         */
-       /* if(!DataSourceUtils.isDbConnected(connection)){
-            connection=DorisUtils.getConnection();
-        }*/
         for (LaneNorm value: laneNorms){
             insertPstmt.setString(1, value.getCorssId());
             insertPstmt.setString(2, value.getLaneId());

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

@@ -12,7 +12,7 @@ import java.sql.SQLException;
  **/
 public class DataSourceUtils {
 
-    public static boolean isDbConnected(Connection con) {
+    public static boolean  isDbConnected(Connection con) {
         //final String CHECK_SQL_QUERY = "SELECT 1";
         try {
             if(con != null && con.isValid(1000)){