统一执行线程池，调整参数

dapeng · dapeng · commit 16398a1b1ffe · 2020-06-22T20:08:48.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
@@ -67,8 +67,6 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, clickhouseClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("clickhouseAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 
 }
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
@@ -76,9 +76,6 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, db2lientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("dbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
-
     }
 
 }
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
@@ -83,8 +83,6 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, impalaClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("impalaAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 
 
diff --git a/launcher/src/main/test/java/com/dtstack/flink/sql/launcher/PluginLoadModeTest.java b/launcher/src/main/test/java/com/dtstack/flink/sql/launcher/PluginLoadModeTest.java
@@ -50,7 +50,7 @@ public static void testClasspathMode() throws Exception {
 
 
     public static void testRocSql() throws Exception{
-        String[] sql = new String[]{"-mode", "local", "-sql", "/Users/roc/Documents/flink_sql/sql/zy_sql/hbase_side.sql", "-name", "roc",
+        String[] sql = new String[]{"-mode", "local", "-sql", "/Users/roc/Documents/flink_sql/sql/impala_sink.sql", "-name", "roc",
                 "-localSqlPluginPath", "/Users/roc/workspace/git_code/flinkStreamSQL/plugins",
                 "-remoteSqlPluginPath", "/Users/roc/workspace/git_code/flinkStreamSQL/plugins",
                 "-flinkconf", "/Users/roc/Documents/flink_sql/flinkconf",
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
@@ -76,8 +76,6 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("mysqlAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 
 }
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
@@ -68,7 +68,5 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, oracleClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("oracleAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 }
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
@@ -75,7 +75,5 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("polardbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 }
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
@@ -76,8 +76,6 @@ public void open(Configuration parameters) throws Exception {
         vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, pgClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("postgresqlAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 
 }
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
@@ -35,6 +35,7 @@
 import io.vertx.ext.sql.SQLClient;
 import io.vertx.ext.sql.SQLConnection;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
@@ -86,6 +87,15 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     private transient ThreadPoolExecutor executor;
 
+    private final static int MAX_TASK_QUEUE_SIZE = 100000;
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        executor = new ThreadPoolExecutor(MAX_DB_CONN_POOL_SIZE_LIMIT, MAX_DB_CONN_POOL_SIZE_LIMIT, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(MAX_TASK_QUEUE_SIZE), new DTThreadFactory("rdbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy());
+    }
+
     public RdbAsyncReqRow(BaseSideInfo sideInfo) {
         super(sideInfo);
         init(sideInfo);
@@ -248,10 +258,6 @@ 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
@@ -78,7 +78,5 @@ public void open(Configuration parameters) throws Exception {
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, sqlserverClientConfig));
-        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
-                new LinkedBlockingQueue<>(10000), new DTThreadFactory("sqlServerAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -67,8 +67,6 @@ public void open(Configuration parameters) throws Exception {`
`67`	`67`	`vo.setFileResolverCachingEnabled(false);`
`68`	`68`	`Vertx vertx = Vertx.vertx(vo);`
`69`	`69`	`setRdbSqlClient(JDBCClient.createNonShared(vertx, clickhouseClientConfig));`
`70`		`- setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
`71`		`- new LinkedBlockingQueue<>(10000), new DTThreadFactory("clickhouseAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`72`	`70`	`}`
`73`	`71`
`74`	`72`	`}`
Original file line number	Diff line number	Diff line change
`@@ -76,9 +76,6 @@ public void open(Configuration parameters) throws Exception {`
`76`	`76`	`vo.setFileResolverCachingEnabled(false);`
`77`	`77`	`Vertx vertx = Vertx.vertx(vo);`
`78`	`78`	`setRdbSqlClient(JDBCClient.createNonShared(vertx, db2lientConfig));`
`79`		`- setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
`80`		`- new LinkedBlockingQueue<>(10000), new DTThreadFactory("dbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`81`		`-`
`82`	`79`	`}`
`83`	`80`
`84`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,8 +83,6 @@ public void open(Configuration parameters) throws Exception {`
`83`	`83`	`vo.setFileResolverCachingEnabled(false);`
`84`	`84`	`Vertx vertx = Vertx.vertx(vo);`
`85`	`85`	`setRdbSqlClient(JDBCClient.createNonShared(vertx, impalaClientConfig));`
`86`		`- setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
`87`		`- new LinkedBlockingQueue<>(10000), new DTThreadFactory("impalaAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`88`	`86`	`}`
`89`	`87`
`90`	`88`
Original file line number	Diff line number	Diff line change
`@@ -76,8 +76,6 @@ public void open(Configuration parameters) throws Exception {`
`76`	`76`	`vo.setFileResolverCachingEnabled(false);`
`77`	`77`	`Vertx vertx = Vertx.vertx(vo);`
`78`	`78`	`setRdbSqlClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));`
`79`		`- setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
`80`		`- new LinkedBlockingQueue<>(10000), new DTThreadFactory("mysqlAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`81`	`79`	`}`
`82`	`80`
`83`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -68,7 +68,5 @@ public void open(Configuration parameters) throws Exception {`
`68`	`68`	`vo.setFileResolverCachingEnabled(false);`
`69`	`69`	`Vertx vertx = Vertx.vertx(vo);`
`70`	`70`	`setRdbSqlClient(JDBCClient.createNonShared(vertx, oracleClientConfig));`
`71`		`- setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
`72`		`- new LinkedBlockingQueue<>(10000), new DTThreadFactory("oracleAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`73`	`71`	`}`
`74`	`72`	`}`
Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,5 @@ public void open(Configuration parameters) throws Exception {`
`75`	`75`	`vo.setFileResolverCachingEnabled(false);`
`76`	`76`	`Vertx vertx = Vertx.vertx(vo);`
`77`	`77`	`setRdbSqlClient(JDBCClient.createNonShared(vertx, mysqlClientConfig));`
`78`		`- setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
`79`		`- new LinkedBlockingQueue<>(10000), new DTThreadFactory("polardbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`80`	`78`	`}`
`81`	`79`	`}`
Original file line number	Diff line number	Diff line change
`@@ -78,7 +78,5 @@ public void open(Configuration parameters) throws Exception {`
`78`	`78`	`vo.setFileResolverCachingEnabled(false);`
`79`	`79`	`Vertx vertx = Vertx.vertx(vo);`
`80`	`80`	`setRdbSqlClient(JDBCClient.createNonShared(vertx, sqlserverClientConfig));`
`81`		`- setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
`82`		`- new LinkedBlockingQueue<>(10000), new DTThreadFactory("sqlServerAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`83`	`81`	`}`
`84`	`82`	`}`