添加日志

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

@@ -30,8 +30,9 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
     EnvProperties envProps;
     private static transient ExecutorService executorService;
     private static transient DruidDataSource dataSource;
-    private static final int BATCH_SIZE = 1000;
+    private static final int BATCH_SIZE = 2000;
     private static final int FLUSH_INTERVAL = 500;
+    private static final int INIT_CAPACITY = (int)(BATCH_SIZE / 0.75 + 1);
     private ArrayBlockingQueue<String> sqlBatch = new ArrayBlockingQueue<>(BATCH_SIZE * 2);
     private static transient ScheduledExecutorService scheduledExecutorService;
     private AtomicBoolean flushing = new AtomicBoolean(false);
@@ -42,7 +43,7 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
 
     @Override
     public void open(Configuration parameters) throws Exception {
-        executorService = new ThreadPoolExecutor(10, 10, 20, TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000));
+        executorService = new ThreadPoolExecutor(15, 20, 20, TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000));
         // 初始化获取配置
         String configTidbUrl = String.format(envProps.getDb_url(), envProps.getDb_host(), envProps.getDb_port(), envProps.getDb_database());
         dataSource = new DruidDataSource();
@@ -50,8 +51,8 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
         dataSource.setUsername(envProps.getDb_username());
         dataSource.setPassword(envProps.getDb_password());
         dataSource.setUrl(configTidbUrl);
-        dataSource.setMaxActive(30);
-        dataSource.setInitialSize(20);
+        dataSource.setMaxActive(50);
+        dataSource.setInitialSize(30);
         dataSource.setTestWhileIdle(true);
         dataSource.setMaxWait(20000);
         dataSource.setValidationQuery("select 1");
@@ -81,7 +82,7 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
                     flushing.set(false);
                     return;
                 }
-                List<String> batch = new ArrayList<>(1334);
+                List<String> batch = new ArrayList<>(INIT_CAPACITY);
                 sqlBatch.drainTo(batch, BATCH_SIZE);
                 if (batch.isEmpty()) {
                     flushing.set(false);
@@ -103,7 +104,8 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
                                 logger.error("事务回滚异常", ex);
                             }
                             logger.error("------错误时间:{}-------------异常:", new Date(), e);
-                            SqlErrorLog errorLog = new SqlErrorLog(new Date(), String.join(";", batch), e.getMessage());
+                            //SqlErrorLog errorLog = new SqlErrorLog(new Date(), String.join(";", batch), e.getMessage());
+                            SqlErrorLog errorLog = new SqlErrorLog(new Date(), batch.get(0), e.getMessage());
                             writeErrLogDb(errorLog);
                         }
                     } catch (Exception e) {