【logminer优化，增加睡眠时间，减少无效查询】【30802】

Author: kanata163

flinkx-oraclelogminer/flinkx-oraclelogminer-reader/src/main/java/com/dtstack/flinkx/oraclelogminer/format/LogMinerConnection.java

@@ -103,7 +103,7 @@ public class LogMinerConnection {
 
     private long lastQueryTime;
 
-    private static final long QUERY_LOG_INTERVAL = 1000;
+    private static final long QUERY_LOG_INTERVAL = 10000;
 
     private boolean logMinerStarted = false;
 
@@ -151,6 +151,19 @@ public void disConnect() throws SQLException{
     public void startOrUpdateLogMiner(Long startScn) {
         String startSql = null;
         try {
+            // 防止没有数据更新的时候频繁查询数据库，限定查询的最小时间间隔 QUERY_LOG_INTERVAL
+            if (lastQueryTime > 0) {
+                long time = System.currentTimeMillis() - lastQueryTime;
+                if (time < QUERY_LOG_INTERVAL) {
+                    try {
+                        Thread.sleep(QUERY_LOG_INTERVAL-time);
+                    } catch (InterruptedException e) {
+                        LOG.warn("", e);
+                    }
+                }
+            }
+            lastQueryTime = System.currentTimeMillis();
+
             if (logMinerConfig.getSupportAutoAddLog()) {
                 startSql = oracleVersion == 10 ? SqlUtil.SQL_START_LOG_MINER_AUTO_ADD_LOG_10 : SqlUtil.SQL_START_LOG_MINER_AUTO_ADD_LOG;
             } else {

flinkx-oraclelogminer/flinkx-oraclelogminer-reader/src/main/java/com/dtstack/flinkx/oraclelogminer/format/LogMinerListener.java

@@ -28,7 +28,13 @@
 
 import java.util.Collections;
 import java.util.Map;
-import java.util.concurrent.*;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author jiangbo
@@ -99,7 +105,6 @@ public void run() {
                     logMinerConnection.closeStmt();
                     logMinerConnection.startOrUpdateLogMiner(positionManager.getPosition());
                     logMinerConnection.queryData(positionManager.getPosition());
-
                     LOG.debug("Update log and continue read:{}", positionManager.getPosition());
                 }
             } catch (Exception e) {