车道指标增加日志级别配置 方便现场调试程序

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

@@ -16,11 +16,9 @@ import org.apache.flink.table.api.Table;
 import org.apache.flink.util.Collector;
 
 import java.math.BigDecimal;
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.Statement;
+import java.sql.*;
 import java.util.*;
+import java.util.Date;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -42,7 +40,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
     //
     private ValueState<Boolean> processCycle;
 
-    private Connection connection;
+    Connection connection;
     //每个路口允许的最大时间数据
     private static Long maxTravel=50000L;
     //每天的cycleOrder 不允许重复
@@ -78,6 +76,23 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
         }
     }
 
+    /**
+     * 判断jdbc 是否失效 失效重新获取连接
+     * @param con
+     * @return
+     */
+    public  boolean isDbConnected(Connection con) {
+        //final String CHECK_SQL_QUERY = "SELECT 1";
+        try {
+            if(!con.isClosed() || con!=null){
+                return true;
+            }
+        } catch (SQLException e) {
+            return false;
+        }
+        return false;
+    }
+
     @Override
     public void processElement1(Cycle cycle, KeyedCoProcessFunction<String, Cycle, TravelEvent, List<LaneNorm>>.Context ctx, Collector<List<LaneNorm>> out) throws Exception {
         processCycle.update(true);
@@ -90,10 +105,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) {
-            if(connection==null){
+            loggerLevel = applicationUtils.getLoggerLevel();
+            if(isDbConnected(connection)){
                 connection=DorisUtils.getConnection();
             }
-            loggerLevel = applicationUtils.getLoggerLevel();
             //查询相位方向信息
             dict.put("phaseDirection", getDictResult(connection, "SELECT\n" +
                     "  a.ID id,\n" +

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

@@ -52,6 +52,7 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
      */
     @Override
     public void invoke(List<LaneNorm> laneNorms, SinkFunction.Context context) throws Exception {
+
         if(laneNorms.size()==0){
             return;
         }

realtime/hologram-streaming/src/main/resources/doris.properties

 driver=com.mysql.jdbc.Driver
-host=10.243.0.22
+host=localhost
 port=9030
 username=root
 password=mima