增加并行度

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

@@ -31,8 +31,7 @@ public class MysqlDataTransferSink extends RichSinkFunction<String> {
 
     @Override
     public void open(Configuration parameters) throws Exception {
-
-        executorService = new ThreadPoolExecutor(10, 10, 20, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
+        executorService = new ThreadPoolExecutor(6, 6, 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);
@@ -42,7 +41,7 @@ public class MysqlDataTransferSink extends RichSinkFunction<String> {
         dataSource.setPassword(envProps.getDb_password());
         dataSource.setUrl(configTidbUrl);
         dataSource.setMaxActive(30);
-        dataSource.setInitialSize(10);
+        dataSource.setInitialSize(20);
         dataSource.setTestWhileIdle(true);
         dataSource.setMaxWait(20000);
         dataSource.setValidationQuery("select 1");

src/main/java/com/dsk/flink/dsc/sync/SyncCustomerDataSource.java

@@ -48,7 +48,7 @@ public class SyncCustomerDataSource {
     public static void main(String[] args) throws Exception {
 
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-        //env.setParallelism(1);
+        //env.setParallelism(3);
         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(50, 30000));
         env.enableCheckpointing(300000);
         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
@@ -78,7 +78,7 @@ public class SyncCustomerDataSource {
         }
 
         SingleOutputStreamOperator<String> kafkaSource = env.addSource(kafkaConsumer)
-                .setParallelism(1)
+                //.setParallelism(1)
                 .name("kafka-source")
                 .uid("kafka-source");