增加并行度

src/main/java/com/dsk/flink/dsc/common/function/AsyncMysqlDataTransferFunctionNew.java

@@ -36,7 +36,7 @@ public class AsyncMysqlDataTransferFunctionNew extends RichAsyncFunction<JSONObj
     @Override
     public void open(Configuration parameters) throws Exception {
         //初始化线程池
-        executorService = new ThreadPoolExecutor(4 , 4, 20, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
+        executorService = new ThreadPoolExecutor(10 , 10, 20, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
     }
 
     @Override

src/main/java/com/dsk/flink/dsc/common/sink/MysqlDataTransferSink.java

@@ -32,7 +32,7 @@ public class MysqlDataTransferSink extends RichSinkFunction<String> {
     @Override
     public void open(Configuration parameters) throws Exception {
 
-        executorService = new ThreadPoolExecutor(4, 4, 20, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
+        executorService = new ThreadPoolExecutor(10, 10, 20, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
         //初始化获取配置
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         //System.out.println(configTidbUrl);

src/main/java/com/dsk/flink/dsc/utils/EtlUtils.java

@@ -20,6 +20,12 @@ public class EtlUtils {
         properties.setProperty("max.partition.fetch.bytes", "2097152000");
         properties.setProperty("flink.consumer.max.fetch.size", "2097152000");
         properties.setProperty("frame.size", "2097152000");
+        properties.setProperty("session.timeout.ms", "30000");
+        properties.setProperty("heartbeat.interval.ms", "10000");
+        properties.setProperty("request.timeout.ms", "60000");
+        properties.setProperty("retries", "3");
+        properties.setProperty("retry.backoff.ms", "5000");
+
         return properties;
     }