车道指标相关计算逻辑

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

@@ -2,10 +2,7 @@ package com.zhht.irn.sink;
 
 import com.zhht.irn.entity.Cycle;
 import com.zhht.irn.entity.Stage;
-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 com.zhht.irn.utils.*;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 
@@ -47,8 +44,16 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
         if(connection != null) connection.close();
     }
 
+
+
     @Override
     public void invoke(Cycle cycle, Context context) throws Exception {
+        /**
+         * 添加连接判断 防止过长时间没写入数据导致的jdbc 失效
+         */
+        if(DataSourceUtils.isDbConnected(connection)){
+            connection=DorisUtils.getConnection();
+        }
         insertCyclePstmt.setLong(1, cycle.getId());
         insertCyclePstmt.setLong(2, cycle.getCycleOrder());
         insertCyclePstmt.setString(3, cycle.getCrossCode());

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

 package com.zhht.irn.sink;
 
 import com.zhht.irn.entity.LaneNorm;
-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 com.zhht.irn.utils.*;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
@@ -52,10 +49,15 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
      */
     @Override
     public void invoke(List<LaneNorm> laneNorms, SinkFunction.Context context) throws Exception {
-
         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

+package com.zhht.irn.utils;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/**
+ * @Description:
+ * @Author ：Marinh
+ * @Param:
+ * @retrun:
+ * @Creat :2023-01-09-9:47
+ **/
+public class DataSourceUtils {
+
+    public static 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;
+    }
+
+}