[代码修改] 获取数据库连接修改为从druid获取

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

 package com.zhht.irn.functions;
 
+import com.alibaba.druid.pool.DruidPooledConnection;
 import com.zhht.irn.entity.*;
 import com.zhht.irn.enums.EventType;
 import com.zhht.irn.utils.*;
@@ -28,7 +29,7 @@ import java.util.stream.Stream;
  * @retrun:
  * @Creat :2022-11-04-9:53
  **/
-public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm> {
+public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>> {
     // state 管理当前周期中存储的旅行事件
     private MapState<String, List<TravelEvent>> cycleTravel;
     // 临时
@@ -48,9 +49,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     }
 
     @Override
-    public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
+    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) {
@@ -119,13 +120,16 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
             list.add(laneNorm);
             //交通流量为0得不参与计算
             if (laneNorm != null && laneNorm.getTrafficCapacity() != 0d) {
-                out.collect(laneNorm);
+                laneNorms.add(laneNorm);
             }
         }
         //获取所有方向的指标
-        getPositionNorm(list, positionList, cycle, out);
+        getPositionNorm(list, positionList, cycle, laneNorms);
         //获取所有流向的指标
-        getPositionNameNorm(list, positionNameList, cycle, out);
+        getPositionNameNorm(list, positionNameList, cycle, laneNorms);
+        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());
         //插入计算过程中缓存的数据
@@ -138,7 +142,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     }
 
     @Override
-    public void processElement2(TravelEvent event, KeyedCoProcessFunction<String, Cycle, TravelEvent, LaneNorm>.Context ctx, Collector<LaneNorm> out) throws Exception {
+    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);
         }
@@ -159,7 +163,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
 
 
     @Override
-    public void onTimer(long timestamp, OnTimerContext ctx, Collector<LaneNorm> out) throws Exception {
+    public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<LaneNorm>> out) throws Exception {
         super.onTimer(timestamp, ctx, out);
         System.out.println("触发定时器了。。。。");
         //通过定时器清空字典表
@@ -213,9 +217,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
      * @param list
      * @param positionNameList
      * @param cycle
-     * @param out
+     * @param collectList
      */
-    private static void getPositionNameNorm(List<LaneNorm> list, List<Map> positionNameList, Cycle cycle, Collector<LaneNorm> out) {
+    private static void getPositionNameNorm(List<LaneNorm> list, List<Map> positionNameList, Cycle cycle,  List<LaneNorm> collectList) {
         if (list.size() > 0) {
             for (Map map : positionNameList) {
                 String position = String.valueOf(map.get("position"));
@@ -255,7 +259,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
                 laneNorm.setSpotSpaceShare(d > 1d ? 1d : d);
                 laneNorm.setSpotSpaceLength(sum1);
                 //发送方向整体信息
-                out.collect(laneNorm);
+                collectList.add(laneNorm);
             }
         }
     }
@@ -266,9 +270,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
      * @param list
      * @param positionList
      * @param cycle
-     * @param out
+     * @param collectList
      */
-    private static void getPositionNorm(List<LaneNorm> list, List<Map> positionList, Cycle cycle, Collector<LaneNorm> out) {
+    private static void getPositionNorm(List<LaneNorm> list, List<Map> positionList, Cycle cycle, List<LaneNorm> collectList) {
         if (list.size() > 0) {
             for (Map map : positionList) {
                 String position = String.valueOf(map.get("position"));
@@ -306,7 +310,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
                 laneNorm.setSpotSpaceShare(d > 1d ? 1d : d);
                 laneNorm.setSpotSpaceLength(sum1);
                 //发送方向整体信息
-                out.collect(laneNorm);
+                collectList.add(laneNorm);
             }
         }
     }
@@ -494,9 +498,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         return (maxQueneLength * 6);
     }
 
-    private static Connection getConnection() throws Exception {
+    private static DruidPooledConnection getConnection() throws Exception {
         //修改为从连接池中取连接
-        Connection connection =  DruidConnectPoolUtils.getConncent();
+        DruidPooledConnection connection =  DruidConnectPoolUtils.getConnection();
         //Connection connection = DorisUtils.getConnection();
         return connection;
     }

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

@@ -87,16 +87,16 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
         carRecordState = getRuntimeContext().getMapState(carRecordStateDescriptor);
 
 
-        dataSource = DruidConnectPoolUtils.getDataSource
+        connection = DruidConnectPoolUtils.getConnection();
             //    ("jdbc:mysql://localhost:9030/demo?characterEncoding=utf8",
             //    ("jdbc:mysql://localhost:3307/test?characterEncoding=utf8",
             //    ("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8",
-                ("jdbc:mysql://10.243.0.22:9030/demo?characterEncoding=utf8",
+               /* ("jdbc:mysql://10.243.0.22:9030/demo?characterEncoding=utf8",
                         "root",
                 //        "root123");
                         "mima");
-
-        connection = dataSource.getConnection();
+*/
+       // dataSource.getConnection();
         dim_cnt_cross_lane_position = getLaneDimData(connection);
 
     }
@@ -754,7 +754,7 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
      */
     private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
 
-        DruidPooledConnection connection = dataSource.getConnection();
+        DruidPooledConnection connection = DruidConnectPoolUtils.getConnection();
 
         String sql = "INSERT INTO demo.app_cross_line_travel_car_info" +
                 "(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +

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

@@ -77,7 +77,7 @@ public class LaneNormJob {
         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<LaneNorm> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4);
+        SingleOutputStreamOperator<List<LaneNorm>> process = cycleTravelEventConnectedStreams.process(new LaneNormProcessFunction()).setParallelism(4);
         process.addSink(new LaneNormSink()).name("laneNormJobSink");
         env.execute("laneNormJob");
     }

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

@@ -3,6 +3,7 @@ package com.zhht.irn.sink;
 import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
 import com.zhht.irn.utils.DateUtils;
 import com.zhht.irn.utils.DorisUtils;
+import com.zhht.irn.utils.DruidConnectPoolUtils;
 import com.zhht.irn.utils.MySQLUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
@@ -28,7 +29,7 @@ public class DirectionEvalSecondSink extends RichSinkFunction<DirectionEvalSecon
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = DorisUtils.getConnection();
+        connection = DruidConnectPoolUtils.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.getConncent();
+        connection = DruidConnectPoolUtils.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

@@ -12,6 +12,7 @@ import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 
 import java.sql.Connection;
 import java.sql.PreparedStatement;
+import java.util.List;
 
 /**
  * @Description:
@@ -20,7 +21,7 @@ import java.sql.PreparedStatement;
  * @retrun:
  * @Creat :2022-10-31-14:11
  **/
-public class LaneNormSink  extends RichSinkFunction<LaneNorm> {
+public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
 
     Connection connection;
     PreparedStatement insertPstmt;
@@ -31,7 +32,7 @@ public class LaneNormSink  extends RichSinkFunction<LaneNorm> {
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         // path暂定为固定值
-        connection = DruidConnectPoolUtils.getConncent();
+        connection = DruidConnectPoolUtils.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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
@@ -50,33 +51,32 @@ public class LaneNormSink  extends RichSinkFunction<LaneNorm> {
      * 1000w的数据  1000w次
      */
     @Override
-    public void invoke(LaneNorm value, SinkFunction.Context context) throws Exception {
-        insertPstmt.setString(1, value.getCorssId());
-        insertPstmt.setString(2, value.getLaneId());
-        insertPstmt.setString(3, value.getDirection());
-        insertPstmt.setLong(4, value.getCycleOrder());
-        insertPstmt.setDouble(5, value.getMaxQueueLength());
-        insertPstmt.setString(6,value.getEffectGreenTime());
-        insertPstmt.setDouble(7,value.getPassCapacity());
-        insertPstmt.setDouble(8,value.getTrafficCapacity());
-        insertPstmt.setDouble(9,value.getSaturation());
-        insertPstmt.setDouble(10,value.getResidualCapacity());
-        insertPstmt.setDouble(11,value.getSpotSpaceShare());
-        insertPstmt.setString(12,value.getLaneName());
-        insertPstmt.setInt(13,value.getOrder());
-        insertPstmt.setString(14, DateUtils.getTodayTime());
-        insertPstmt.setString(15,value.getType());
-        insertPstmt.setInt(16,value.getSpotSpaceLength());
-        insertPstmt.setString(17,value.getCycleStartTime());
-        insertPstmt.setString(18,value.getCycleEndTime());
-        insertPstmt.setString(19,DateUtils.getTodayTime());
-        insertPstmt.execute();
-        /*if(counter < 10) {
-            counter ++;
+    public void invoke(List<LaneNorm> laneNorms, SinkFunction.Context context) throws Exception {
+        if(laneNorms.size()==0){
+            return;
+        }
+        for (LaneNorm value: laneNorms){
+            insertPstmt.setString(1, value.getCorssId());
+            insertPstmt.setString(2, value.getLaneId());
+            insertPstmt.setString(3, value.getDirection());
+            insertPstmt.setLong(4, value.getCycleOrder());
+            insertPstmt.setDouble(5, value.getMaxQueueLength());
+            insertPstmt.setString(6,value.getEffectGreenTime());
+            insertPstmt.setDouble(7,value.getPassCapacity());
+            insertPstmt.setDouble(8,value.getTrafficCapacity());
+            insertPstmt.setDouble(9,value.getSaturation());
+            insertPstmt.setDouble(10,value.getResidualCapacity());
+            insertPstmt.setDouble(11,value.getSpotSpaceShare());
+            insertPstmt.setString(12,value.getLaneName());
+            insertPstmt.setInt(13,value.getOrder());
+            insertPstmt.setString(14, DateUtils.getTodayTime());
+            insertPstmt.setString(15,value.getType());
+            insertPstmt.setInt(16,value.getSpotSpaceLength());
+            insertPstmt.setString(17,value.getCycleStartTime());
+            insertPstmt.setString(18,value.getCycleEndTime());
+            insertPstmt.setString(19,DateUtils.getTodayTime());
             insertPstmt.addBatch();
-        } else {
-            insertPstmt.executeBatch();
-            counter = 0;
-        }*/
+        }
+        insertPstmt.executeBatch();
     }
 }

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

@@ -2,6 +2,7 @@ package com.zhht.irn.sink;
 
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.utils.DorisUtils;
+import com.zhht.irn.utils.DruidConnectPoolUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 
@@ -23,7 +24,7 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
     @Override
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
-        connection = DorisUtils.getConnection();
+        connection = DruidConnectPoolUtils.getConnection();
 
         //目前值考虑插入，不考虑更新
         String insertSql="" +

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

 package com.zhht.irn.utils;
 
 import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.pool.DruidPooledConnection;
 import org.apache.flink.api.java.utils.ParameterTool;
 
 import java.io.InputStream;
@@ -55,7 +56,7 @@ public class DruidConnectPoolUtils {
         pool=dataSource;
     }
 
-    public static Connection getConncent(){
+    public static DruidPooledConnection getConnection(){
         try {
             if(pool==null){
                 InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("doris.properties");