2022-12-23 更改查询维表连接方式

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

@@ -5,6 +5,7 @@ import com.alibaba.druid.pool.DruidPooledConnection;
 
 
 import com.zhht.irn.entity.dto.*;
+import com.zhht.irn.job.TravelSituationAnalysisJobGetCycleData;
 import com.zhht.irn.utils.DruidConnectPoolUtils;
 import org.apache.flink.api.common.state.*;
 import org.apache.flink.api.common.time.Time;
@@ -12,6 +13,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
 import org.apache.flink.util.Collector;
+import org.apache.log4j.Logger;
 
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
@@ -27,6 +29,8 @@ import java.util.*;
  **/
 public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>> {
 
+    private static Logger logger = Logger.getLogger(TravelCarInfoCoProcessFunction.class);
+
     //定义状态，用于存储旅行信息，当周期信号数据过来的时候，触发计算
     //状态数据存储结构为   Map（路口id--》（Map（车道id-》（List（旅行信息））））） 一辆车只有一条旅行信息
     // 这样就把每个路口的数据进行分流，同时，对每个路口的数据，也按照了不同的车道进行了分流，同一个路口，同一个车道的数据存在一个list中
@@ -37,8 +41,6 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     private  MapState<Integer, CycleSignalData> alreadyHandleCycleSignalDataState;  //已经处理过的周期数据，避免重复触发
 
     private ListState<Long> cycleSectionState ;
-
-    private transient DruidPooledConnection connection;
     private static Map<String, Map<String, String>> dim_cnt_cross_lane_position;
 
     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@@ -73,29 +75,18 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                 .build();
         cycleSignalDataStateDescriptor.enableTimeToLive(stateTtlConfig4CycleData);
 
-
-
         travelInfoState = getRuntimeContext().getMapState(travelInfoStateDescriptor);
         cycleSignalDataState = getRuntimeContext().getMapState(cycleSignalDataStateDescriptor);
         alreadyHandleCycleSignalDataState = getRuntimeContext().getMapState(alreadyHandleCycleSignalDataStateDescriptor);
         cycleSectionState = getRuntimeContext().getListState(cycleSectionStateDescriptor);
-         connection = DruidConnectPoolUtils.getConnection();
-/*     //   connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://localhost:3309/test?characterEncoding=utf8",
-        connection = DruidConnectPoolUtils.getDataSource("jdbc:mysql://10.243.0.26:3306/test?characterEncoding=utf8",
-
-                "root","mima").getConnection();*/
 
-        dim_cnt_cross_lane_position = getLaneDimData(connection);
+        dim_cnt_cross_lane_position = getLaneDimData();
 
     }
 
     @Override
     public void close() throws Exception {
         super.close();
-        if(connection!=null){
-            connection.close();
-        }
-
     }
 
     //把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
@@ -104,7 +95,9 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     public void processElement1(TravelInfo value, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.Context ctx,
                                 Collector<List<TravelLineSinkInfo>> out) throws Exception {
 
-        System.out.println("路口id"+value.getCrossId() +"车辆id"+value.getCarId() + "进入路口时间："+sdf.format(value.getInCrossTime()));
+       // 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()));
+
         String crossId = value.getCrossId();
         Iterator<Map<String, List<TravelInfo>>> iterator = travelInfoState.values().iterator();
         //如果状态为空，则进行初始化
@@ -135,8 +128,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
         //当前路口有数据，而且 当前车道 也有数据
         if(crossListMap != null && crossListMap.containsKey(value.getInCrossLineId()))  {
             List<TravelInfo> lineTravelInfos = crossListMap.get(value.getInCrossLineId());
-            //如果，当前车道的数据，积压超过2000条，则进行清理操作
-            if(lineTravelInfos.size()>10000){
+            //如果，当前车道的数据，积压超过5000条，则进行清理操作
+            if(lineTravelInfos.size()>5000){
                 System.out.println("移除积压的数据。。。。");
                 List<TravelInfo> travelInfoList = orderByInCrossTime(lineTravelInfos);
                 List<TravelInfo> travelInfoListNew = travelInfoList.subList(500, travelInfoList.size());
@@ -159,7 +152,9 @@ 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()));
+
             //判断当前周期是否执行过了,存在周期数据重复下发的情况，只触发一次计算
             if (alreadyHandleCycleSignalDataState.contains(value.getCycleOrder())) {
                 System.out.println("周期数据重复----当前路口id是" + value.getCrossCode() + "当前周期已经计算完成了" + value.getCycleOrder());
@@ -228,7 +223,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     public void onTimer(long timestamp, CoProcessFunction<TravelInfo, CycleSignalData, List<TravelLineSinkInfo>>.OnTimerContext ctx, Collector<List<TravelLineSinkInfo>> out) throws Exception {
         super.onTimer(timestamp, ctx, out);
 
-        dim_cnt_cross_lane_position = getLaneDimData(connection);
+        dim_cnt_cross_lane_position = getLaneDimData();
     }
 
     /**
@@ -347,11 +342,6 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
 
     }
 
-
-
-
-
-
     /**
      * 过滤出旅行信息状态中，属于这个周期内的计算数据，返回，对于已经过期的数据进行清理
      * @param beginDateTime
@@ -458,15 +448,15 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
     /**
      * 查询车道维表数据-数据存在map中，key是路口编号，
      *
-     *
-     * @param connection
      * @return
      * @throws SQLException
      */
-    private Map<String, Map<String, String>> getLaneDimData(DruidPooledConnection connection) throws SQLException {
+    private Map<String, Map<String, String>> getLaneDimData() throws SQLException {
+
 
+        DruidPooledConnection connection1 = DruidConnectPoolUtils.getConnection();
         String sql = "select cross_id, lane_id,position,turn_direction as flow_direction from dim_cnt_cross_lane_position  ";
-        PreparedStatement ps = connection.prepareStatement(sql);
+        PreparedStatement ps = connection1.prepareStatement(sql);
         ResultSet resultSet = ps.executeQuery();
 
         Map<String, Map<String, String>> dimData = new HashMap<>();
@@ -484,6 +474,11 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
                 dimData.put(resultSet.getString("cross_id"), map);
             }
         }
+
+        if(connection1!=null){
+            connection1.close();
+        }
+
         return dimData;
 
     }
@@ -753,7 +748,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
      * 将每个车道流向预聚合的数据存入mysql中去
      * @param list
      * @throws SQLException
-     */
+     *//*
     private void sinkTravelLineSinkInfoToMysql(List<TravelLineSinkInfo> list) throws SQLException {
 
 
@@ -798,6 +793,6 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
             }
             preparedStatement.executeBatch();
 
-    }
+    }*/
 
 }

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

@@ -22,8 +22,7 @@ import org.apache.flink.streaming.api.environment.CheckpointConfig;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+
 
 import java.util.List;
 import java.util.Properties;
@@ -54,7 +53,7 @@ public class TravelSituationAnalysisJob {
             env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 
             //设置重启策略
-            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Time.seconds(10)));
+            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
 
 
 

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

@@ -2,9 +2,11 @@ package com.zhht.irn.sink;
 
 import com.alibaba.druid.pool.DruidPooledConnection;
 import com.zhht.irn.entity.dto.TravelLineSinkInfo;
+import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
 import com.zhht.irn.utils.DruidConnectPoolUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.log4j.Logger;
 
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -21,6 +23,8 @@ import java.util.List;
  */
 public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo>> {
 
+    private static Logger logger = Logger.getLogger(TravelLaneInfoSink.class);
+
     Connection connection;
 
 
@@ -28,7 +32,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
     public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         connection = DruidConnectPoolUtils.getConnection();
-        System.out.println("建立数据库连接成功。。。。。。。");
+        logger.info("建立数据库连接成功。。。。。。。");
 
     }
 
@@ -86,8 +90,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
         }
         int[] ints = preparedStatement.executeBatch();
 
-        System.out.println("写入数据成功。。。。。"+ints.length+"...");
-
+        logger.info("写入数据成功。。。。。"+ints.length+"...");
 
     }
 }