车道指标相关计算逻辑

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

@@ -39,6 +39,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     //
     private ValueState<Boolean> processCycle;
 
+    private Connection connection;
+
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
@@ -46,16 +48,26 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         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();
+    }
+
+    @Override
+    public void close() throws Exception {
+        super.close();
+        if(connection!=null){
+            connection.close();
+        }
     }
 
     @Override
     public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
         processCycle.update(true);
         List<LaneNorm> laneNorms = new ArrayList<>();
-
         //初始化字典数据
         if (dict.get("phaseDirection") == null || dict.get("laneList") == null ||  dict.get("dictResult") == null ||  dict.get("positionList") == null ||  dict.get("positionNameList") == null) {
-            Connection connection = getConnection();
+            if(connection==null){
+                connection=DorisUtils.getConnection();
+            }
             //查询相位方向信息
             dict.put("phaseDirection", getDictResult(connection, "SELECT\n" +
                     "  a.ID id,\n" +
@@ -89,10 +101,6 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
             //12小时触发一次
             ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+12*60*60*1000);
             //ctx.timerService().registerProcessingTimeTimer(currentProcessingTime+1*60*1000);
-            //关闭mysql连接
-            if (connection != null) {
-                connection.close();
-            }
         }
         Cycle f1 = cycle;
         // 代表周期已到，需要进行相应的指标计算了

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

@@ -83,8 +83,6 @@ public class LaneNormJob {
             env.execute("laneNormJob");
         }catch (Exception e){
             System.out.println("程序异常关闭->"+e);
-            //程序异常 关闭数据库连接池
-            DruidConnectPoolUtils.colseDataSource();
         }
 
     }

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

@@ -29,7 +29,7 @@ public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecon
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = DruidConnectPoolUtils.getConnection();
+        connection = DorisUtils.getConnection();
         String insertSql="insert into app_direction_eval_second_metric" +
                 "(cross_id, lane_id, cycle_order, current_second, acc_arrived_flow, acc_queue_length, acc_stop_count," +
                 " stop_flag_count, statistic_time, cycle_start_time, cycle_end_time, record_date) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

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

@@ -31,7 +31,7 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
         super.open(parameters);
         // path暂定为固定值
         //String path = ".\\realtime\\hologram-streaming\\src\\main\\resources\\doris.properties";
-        connection = DruidConnectPoolUtils.getConnection();
+        connection = DorisUtils.getConnection();
         String cycleSql="insert into app_cycle_green_ratio" +
                 "(id,cycle_order,cross_id,signal_code,begin_time,end_time,duration,green_ratio,loss_time,statistic_time,record_date) " +
                 "values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

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

@@ -32,7 +32,7 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         // path暂定为固定值
-        connection = DruidConnectPoolUtils.getConnection();
+        connection = DorisUtils.getConnection();
         insertPstmt = connection.prepareStatement("insert into " +
                 "app_cross_dimension_norm(cross_id,lane_id,direction,cycle_order,max_queue_length,effect_green_time,pass_capacity,traffic_capacity,saturation,residual_capacity," +
                 "spot_space_share,dimension_name,`order`,statistic_time,`type`,spot_space_length,cycle_start_time,cycle_end_time,record_date) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");

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

@@ -24,7 +24,7 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = DruidConnectPoolUtils.getConnection();
+        connection = DorisUtils.getConnection();
 
         //目前值考虑插入，不考虑更新
         String insertSql="" +

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

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