车道指标相关计算逻辑

realtime/hologram-streaming/pom.xml

@@ -13,13 +13,11 @@
     <version>1.0-SNAPSHOT</version>
 
     <dependencies>
-
         <!-- flink 相关的依赖包 -->
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
@@ -28,19 +26,15 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-clients_${scala.binary.version}</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-core</artifactId>
         </dependency>
-
         <!-- 工具包 -->
-
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
@@ -85,20 +79,15 @@
             <artifactId>flink-table-api-java-bridge_2.11</artifactId>
             <version>1.12.0</version>
         </dependency>
-
         <!-- 导入 druid 的 jar 包 -->
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>druid</artifactId>
             <version>1.2.8</version>
         </dependency>
-
     </dependencies>
-
     <build>
-
         <plugins>
-
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-shade-plugin</artifactId>

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

@@ -106,7 +106,7 @@ 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) {
             loggerLevel = applicationUtils.getLoggerLevel();
-            if(isDbConnected(connection)){
+            if(!DataSourceUtils.isDbConnected(connection)){
                 connection=DorisUtils.getConnection();
             }
             //查询相位方向信息

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

@@ -51,7 +51,7 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
         /**
          * 添加连接判断 防止过长时间没写入数据导致的jdbc 失效
          */
-        if(DataSourceUtils.isDbConnected(connection)){
+        if(!DataSourceUtils.isDbConnected(connection)){
             connection=DorisUtils.getConnection();
         }
         insertCyclePstmt.setLong(1, cycle.getId());

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

@@ -55,7 +55,7 @@ public class LaneNormSink  extends RichSinkFunction<List<LaneNorm>> {
         /**
          * 添加连接判断 防止jdbc 失效
          */
-        if(DataSourceUtils.isDbConnected(connection)){
+        if(!DataSourceUtils.isDbConnected(connection)){
             connection=DorisUtils.getConnection();
         }
         for (LaneNorm value: laneNorms){