修复异步阻塞问题

dapeng · dapeng · commit 46598c486160 · 2020-05-18T10:43:31.000+08:00
diff --git a/clickhouse/clickhouse-side/clickhouse-async-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAsyncReqRow.java b/clickhouse/clickhouse-side/clickhouse-async-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAsyncReqRow.java
@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.side.clickhouse;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -32,6 +33,9 @@
 import org.apache.flink.configuration.Configuration;
 
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 
 public class ClickhouseAsyncReqRow extends RdbAsyncReqRow {
@@ -63,6 +67,8 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, clickhouseClientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("clickhouseAsyncExec")));
     }
 
 }
diff --git a/db2/db2-side/db2-async-side/src/main/java/com/dtstack/flink/sql/side/db2/Db2AsyncReqRow.java b/db2/db2-side/db2-async-side/src/main/java/com/dtstack/flink/sql/side/db2/Db2AsyncReqRow.java
@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.side.db2;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -33,6 +34,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Reason:
@@ -75,6 +79,8 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, db2lientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("dbAsyncExec")));
     }
 
 }
diff --git a/impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java b/impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java
@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.side.impala;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -35,6 +36,9 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Date: 2019/11/12
@@ -79,6 +83,8 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, impalaClientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("impalaAsyncExec")));
     }
 
 
diff --git a/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java b/mysql/mysql-side/mysql-async-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAsyncReqRow.java
@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.side.mysql;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -32,6 +33,9 @@
 import org.apache.flink.configuration.Configuration;
 
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Mysql dim table
@@ -72,6 +76,8 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("mysqlAsyncExec")));
     }
 
 }
diff --git a/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java b/oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java
@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.side.oracle;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -32,6 +33,9 @@
 import org.apache.flink.configuration.Configuration;
 
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 
 public class OracleAsyncReqRow extends RdbAsyncReqRow {
@@ -64,5 +68,7 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, oracleClientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("oracleAsyncExec")));
     }
 }
diff --git a/polardb/polardb-side/polardb-async-side/src/main/java/com/dtstack/flink/sql/side/polardb/PolardbAsyncReqRow.java b/polardb/polardb-side/polardb-async-side/src/main/java/com/dtstack/flink/sql/side/polardb/PolardbAsyncReqRow.java
@@ -17,6 +17,7 @@
  */
 package com.dtstack.flink.sql.side.polardb;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -32,6 +33,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Date: 2019/12/20
@@ -71,5 +75,7 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("polardbAsyncExec")));
     }
 }
diff --git a/postgresql/postgresql-side/postgresql-async-side/src/main/java/com/dtstack/flink/sql/side/postgresql/PostgresqlAsyncReqRow.java b/postgresql/postgresql-side/postgresql-async-side/src/main/java/com/dtstack/flink/sql/side/postgresql/PostgresqlAsyncReqRow.java
@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.side.postgresql;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
@@ -34,6 +35,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Date: 2019-08-11
@@ -72,6 +76,8 @@ public void open(Configuration parameters) throws Exception {
         vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, pgClientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("postgresqlAsyncExec")));
     }
 
 }
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java
@@ -20,6 +20,7 @@
 package com.dtstack.flink.sql.side.rdb.async;
 
 import com.dtstack.flink.sql.enums.ECacheContentType;
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.BaseAsyncReqRow;
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.CacheMissVal;
@@ -46,8 +47,7 @@
 import java.time.Instant;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -84,6 +84,8 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     private AtomicBoolean connectionStatus = new AtomicBoolean(true);
 
+    private transient ThreadPoolExecutor executor;
+
     public RdbAsyncReqRow(BaseSideInfo sideInfo) {
         super(sideInfo);
         init(sideInfo);
@@ -112,18 +114,7 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, CRow input, Resul
             Thread.sleep(100);
         }
         Map<String, Object> params = formatInputParam(inputParams);
-        rdbSqlClient.getConnection(conn -> {
-            if(conn.failed()){
-                connectionStatus.set(false);
-                connectWithRetry(params, input, resultFuture, rdbSqlClient);
-                return;
-            }
-            connectionStatus.set(true);
-            ScheduledFuture<?> timerFuture = registerTimer(input, resultFuture);
-            cancelTimerWhenComplete(resultFuture, timerFuture);
-            handleQuery(conn.result(), params, input, resultFuture);
-        });
-
+        executor.execute(() -> connectWithRetry(params, input, resultFuture, rdbSqlClient));
     }
 
     private void connectWithRetry(Map<String, Object> inputParams, CRow input, ResultFuture<CRow> resultFuture, SQLClient rdbSqlClient) {
@@ -134,6 +125,7 @@ private void connectWithRetry(Map<String, Object> inputParams, CRow input, Resul
             rdbSqlClient.getConnection(conn -> {
                 try {
                     if(conn.failed()){
+                        connectionStatus.set(false);
                         if(failCounter.getAndIncrement() % 1000 == 0){
                             LOG.error("getConnection error", conn.cause());
                         }
@@ -210,12 +202,20 @@ public void close() throws Exception {
             rdbSqlClient.close();
         }
 
+        if(executor != null){
+            executor.shutdown();
+        }
+
     }
 
     public void setRdbSqlClient(SQLClient rdbSqlClient) {
         this.rdbSqlClient = rdbSqlClient;
     }
 
+    public void setExecutor(ThreadPoolExecutor executor) {
+        this.executor = executor;
+    }
+
     private void handleQuery(SQLConnection connection, Map<String, Object> inputParams, CRow input, ResultFuture<CRow> resultFuture){
         String key = buildCacheKey(inputParams);
         JsonArray params = new JsonArray(Lists.newArrayList(inputParams.values()));
diff --git a/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java b/sqlserver/sqlserver-side/sqlserver-async-side/src/main/java/com/dtstack/flink/sql/side/sqlserver/SqlserverAsyncReqRow.java
@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.side.sqlserver;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
@@ -34,6 +35,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Date: 2019/11/26
@@ -74,5 +78,7 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, sqlserverClientConfig));
+        setExecutor(new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10), new DTThreadFactory("sqlServerAsyncExec")));
     }
 }
