车道指标相关计算逻辑

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

@@ -48,6 +48,12 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
 
     @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

@@ -52,6 +52,12 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
         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());