优化

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(10 , 10, 20, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
+        executorService = new ThreadPoolExecutor(12 , 12, 20, TimeUnit.MINUTES, new LinkedBlockingDeque<>());
     }
 
     @Override

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

@@ -22,8 +22,8 @@ public class MysqlDataTransferSink extends RichSinkFunction<String> {
 
     static Logger logger = LoggerFactory.getLogger(MysqlDataTransferSink.class);
     EnvProperties envProps;
-    ExecutorService executorService;
-    DruidDataSource dataSource;
+    private static transient ExecutorService executorService;
+    private static transient DruidDataSource dataSource;
 
     public MysqlDataTransferSink(EnvProperties envProps) {
         this.envProps = envProps;
@@ -68,8 +68,7 @@ public class MysqlDataTransferSink extends RichSinkFunction<String> {
             pt = connection.prepareStatement(sql);
             pt.execute();
         } catch (Exception e) {
-            System.out.println("sql报错----->" + sql);
-            logger.error("------错误时间:{}-----,sql:{}--------异常:{}", DateUtil.now(),sql,e.getMessage());
+            //logger.error("------错误时间:{}-----,sql:{}--------异常:{}", DateUtil.now(),sql,e.getMessage());
             logger.error("异常信息：",e);
             SqlErrorLog errorLog = new SqlErrorLog(new Date(), sql, e.getMessage());
             try {

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

@@ -78,7 +78,7 @@ public class SyncCustomerDataSource {
         }
 
         SingleOutputStreamOperator<String> kafkaSource = env.addSource(kafkaConsumer)
-                //.setParallelism(1)
+                .setParallelism(1)
                 .name("kafka-source")
                 .uid("kafka-source");
 
@@ -99,11 +99,11 @@ public class SyncCustomerDataSource {
         SingleOutputStreamOperator<Tuple3<JSONObject, String, Long>> tsGroupStream = canalJsonStream.map(new CanalMapToTsGroupFunction());
 
         SingleOutputStreamOperator<JSONObject> process = tsGroupStream.keyBy(x -> x.f1)
-                .window(TumblingProcessingTimeWindows.of(Time.milliseconds(300)))
+                .window(TumblingProcessingTimeWindows.of(Time.milliseconds(100)))
                 .process(new GroupTsProcessWindowFunction());
 
         SingleOutputStreamOperator<Tuple3<String, String, Long>> sqlResultStream1 = AsyncDataStream.orderedWait(process,
-                        new AsyncMysqlDataTransferFunctionNew(envProps), 300L, TimeUnit.SECONDS)
+                        new AsyncMysqlDataTransferFunctionNew(envProps), 120L, TimeUnit.SECONDS)
                 .filter(new FilterFunction<Tuple3<String, String, Long>>() {
                     @Override
                     public boolean filter(Tuple3<String, String, Long> value) throws Exception {
@@ -116,7 +116,7 @@ public class SyncCustomerDataSource {
         //sqlResultStream1.print("async sql==>");
 
         SingleOutputStreamOperator<String> groupWindowSqlResultStream = sqlResultStream1.keyBy(value -> value.f1)
-                .window(TumblingProcessingTimeWindows.of(Time.milliseconds(300)))
+                .window(TumblingProcessingTimeWindows.of(Time.milliseconds(100)))
                 .process(new ProcessWindowFunction<Tuple3<String, String, Long>, String, String, TimeWindow>() {
                     @Override
                     public void process(String s, ProcessWindowFunction<Tuple3<String, String, Long>, String, String,

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

@@ -16,15 +16,17 @@ public class EtlUtils {
         properties.setProperty("sasl.jaas.config", getSaslJaasConfig(username,password));
         properties.setProperty("security.protocol", "SASL_PLAINTEXT");
         properties.setProperty("sasl.mechanism", "SCRAM-SHA-512");
-        properties.setProperty("fetch.max.bytes", "2097152000");
-        properties.setProperty("max.partition.fetch.bytes", "2097152000");
-        properties.setProperty("flink.consumer.max.fetch.size", "2097152000");
-        properties.setProperty("frame.size", "2097152000");
+        properties.setProperty("fetch.max.bytes", "10485760"); //10M
+        //properties.setProperty("max.partition.fetch.bytes", "104857600");
+        properties.setProperty("flink.consumer.max.fetch.size", "5242880");//5m
+        //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");
+        properties.setProperty("receive.buffer.bytes", "2621440"); //2.5m
+        properties.setProperty("max.poll.records", "50");
 
         return properties;
     }