fix 线程数和队列大小

dapeng · dapeng · commit eb0ae659caf4 · 2020-05-28T15:00:19.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,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")));
+        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,8 +76,9 @@ 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")));
+        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,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")));
+        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10000), new DTThreadFactory("impalaAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 
 
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,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")));
+        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,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")));
+        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,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")));
+        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,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")));
+        setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(10000), new DTThreadFactory("postgresqlAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));
     }
 
 }
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,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")));
+        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,8 @@ 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(1, 1, 0, TimeUnit.MILLISECONDS,`
`71`		`- new LinkedBlockingQueue<>(10), new DTThreadFactory("clickhouseAsyncExec")));`
	`70`	`+ setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
	`71`	`+ new LinkedBlockingQueue<>(10000), new DTThreadFactory("clickhouseAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`72`	`72`	`}`
`73`	`73`
`74`	`74`	`}`
Original file line number	Diff line number	Diff line change
`@@ -76,8 +76,9 @@ 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(1, 1, 0, TimeUnit.MILLISECONDS,`
`80`		`- new LinkedBlockingQueue<>(10), new DTThreadFactory("dbAsyncExec")));`
	`79`	`+ setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
	`80`	`+ new LinkedBlockingQueue<>(10000), new DTThreadFactory("dbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
	`81`	`+`
`81`	`82`	`}`
`82`	`83`
`83`	`84`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,8 +83,8 @@ 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(1, 1, 0, TimeUnit.MILLISECONDS,`
`87`		`- new LinkedBlockingQueue<>(10), new DTThreadFactory("impalaAsyncExec")));`
	`86`	`+ setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
	`87`	`+ new LinkedBlockingQueue<>(10000), new DTThreadFactory("impalaAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`88`	`88`	`}`
`89`	`89`
`90`	`90`
Original file line number	Diff line number	Diff line change
`@@ -76,8 +76,8 @@ 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(1, 1, 0, TimeUnit.MILLISECONDS,`
`80`		`- new LinkedBlockingQueue<>(10), new DTThreadFactory("mysqlAsyncExec")));`
	`79`	`+ setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
	`80`	`+ new LinkedBlockingQueue<>(10000), new DTThreadFactory("mysqlAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`81`	`81`	`}`
`82`	`82`
`83`	`83`	`}`
Original file line number	Diff line number	Diff line change
`@@ -68,7 +68,7 @@ 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(1, 1, 0, TimeUnit.MILLISECONDS,`
`72`		`- new LinkedBlockingQueue<>(10), new DTThreadFactory("oracleAsyncExec")));`
	`71`	`+ setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
	`72`	`+ new LinkedBlockingQueue<>(10000), new DTThreadFactory("oracleAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`73`	`73`	`}`
`74`	`74`	`}`
Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ 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(1, 1, 0, TimeUnit.MILLISECONDS,`
`79`		`- new LinkedBlockingQueue<>(10), new DTThreadFactory("polardbAsyncExec")));`
	`78`	`+ setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
	`79`	`+ new LinkedBlockingQueue<>(10000), new DTThreadFactory("polardbAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`80`	`80`	`}`
`81`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -78,7 +78,7 @@ 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(1, 1, 0, TimeUnit.MILLISECONDS,`
`82`		`- new LinkedBlockingQueue<>(10), new DTThreadFactory("sqlServerAsyncExec")));`
	`81`	`+ setExecutor(new ThreadPoolExecutor(50, 50, 0, TimeUnit.MILLISECONDS,`
	`82`	`+ new LinkedBlockingQueue<>(10000), new DTThreadFactory("sqlServerAsyncExec"), new ThreadPoolExecutor.CallerRunsPolicy()));`
`83`	`83`	`}`
`84`	`84`	`}`