[feat] mysql consumer

FlechazoW · FlechazoW · commit 1ed345a16a57 · 2020-09-14T11:59:17.000+08:00
diff --git a/core/src/main/java/com/dtstack/flink/sql/dirtyManager/consumer/AbstractDirtyDataConsumer.java b/core/src/main/java/com/dtstack/flink/sql/dirtyManager/consumer/AbstractDirtyDataConsumer.java
@@ -23,7 +23,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
-import java.sql.SQLException;
 import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -38,8 +37,8 @@ public abstract class AbstractDirtyDataConsumer implements Runnable, Serializabl
 
     protected static final Logger LOG = LoggerFactory.getLogger(AbstractDirtyDataConsumer.class);
 
-    protected long errorLimit = 1000;
-    protected long errorCount = 0;
+    protected Long errorLimit = 1000L;
+    protected Long errorCount = 0L;
 
     protected long count = 0;
 
@@ -49,6 +48,8 @@ public abstract class AbstractDirtyDataConsumer implements Runnable, Serializabl
 
     /**
      * 消费队列数据
+     *
+     * @throws Exception throw exception
      */
     public abstract void consume() throws Exception;
 
@@ -59,6 +60,9 @@ public abstract class AbstractDirtyDataConsumer implements Runnable, Serializabl
 
     /**
      * 初始化消费者，初始化定时任务
+     *
+     * @param properties 任务参数
+     * @throws Exception throw exception
      */
     public abstract void init(Map<String, String> properties) throws Exception;
 
@@ -77,9 +81,9 @@ public void run() {
                 consume();
             }
         } catch (Exception e) {
-            LOG.error("consume dirtyData error");
+            LOG.error("consume dirtyData error", e);
             errorCount++;
-            if (errorCount == errorLimit) {
+            if (errorCount.equals(errorLimit)) {
                 throw new RuntimeException("脏数据消费失败达到上限，任务失败");
             }
         }
diff --git a/core/src/main/java/com/dtstack/flink/sql/dirtyManager/entity/DirtyDataEntity.java b/core/src/main/java/com/dtstack/flink/sql/dirtyManager/entity/DirtyDataEntity.java
@@ -94,4 +94,12 @@ public String toString() {
                 ", field='" + field + '\'' +
                 '}';
     }
+
+    /**
+     * 获取脏数据信息，返回字符数组
+     * @return 脏数据信息字符数组
+     */
+    public String[] get() {
+        return new String[] {dirtyData, String.valueOf(processDate), cause, field};
+    }
 }
diff --git a/core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/DirtyDataManager.java b/core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/DirtyDataManager.java
@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.dirtyManager.consumer.AbstractDirtyDataConsumer;
 import com.dtstack.flink.sql.dirtyManager.entity.DirtyDataEntity;
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.util.PluginUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,6 +31,7 @@
 import java.lang.reflect.Constructor;
 import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -39,7 +41,7 @@
  * Date 2020/8/27 星期四
  */
 public class DirtyDataManager implements Serializable {
-    //TODO 需要确保产生脏数据后至少一次是成功发送给了consumer，至于consumer是否成功消费，manager不需要关心
+
     private static final long serialVersionUID = 7190970299538893497L;
 
     private static final Logger LOG = LoggerFactory.getLogger(DirtyDataManager.class);
@@ -72,6 +74,14 @@ public class DirtyDataManager implements Serializable {
 
     public static AbstractDirtyDataConsumer consumer;
 
+    private static ThreadPoolExecutor dirtyDataConsumer;
+
+    public final static int MAX_POOL_SIZE_LIMIT = 5;
+
+    private final static int MAX_TASK_QUEUE_SIZE = 100;
+
+    private final static String DEFAULT_TYPE = "console";
+
     /**
      * 通过参数生成manager实例，并同时将consumer实例化
      */
@@ -80,17 +90,19 @@ public static DirtyDataManager newInstance(Map<String, String> properties) throw
         manager.blockingInterval = Long.parseLong(properties.getOrDefault("blockingInterval", "60"));
         consumer = createConsumer(properties);
         consumer.init(properties);
-        //TODO 使用线程池创建线程，不要用thread
-        Thread dirtyDataConsumer = new Thread(consumer.setQueue(manager.queue), "dirtyData Consumer");
-        dirtyDataConsumer.start();
+        consumer.setQueue(manager.queue);
+        dirtyDataConsumer = new ThreadPoolExecutor(MAX_POOL_SIZE_LIMIT, MAX_POOL_SIZE_LIMIT, 0, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>(MAX_TASK_QUEUE_SIZE), new DTThreadFactory("dirtyDataConsumer"), new ThreadPoolExecutor.CallerRunsPolicy());
+        dirtyDataConsumer.execute(consumer);
+
         return manager;
     }
 
     /**
      * 通过动态加载的方式加载Consumer
      */
     private static AbstractDirtyDataConsumer createConsumer(Map<String, String> properties) throws Exception {
-        String type = properties.getOrDefault("type", "print");
+        String type = properties.getOrDefault("type", DEFAULT_TYPE);
         String consumerType = DIRTY_CONSUMER_PATH + File.separator + type;
         String consumerJar = PluginUtil.getJarFileDirPath(consumerType, properties.getOrDefault("pluginPath", null), "shipfile");
         String className = CLASS_PRE_STR + "." + type.toLowerCase() + "." + upperCaseFirstChar(type + CLASS_POST_STR);
@@ -106,6 +118,7 @@ private static AbstractDirtyDataConsumer createConsumer(Map<String, String> prop
      * 脏数据收集任务停止，任务停止之前，需要将队列中所有的数据清空
      */
     public void close() throws Exception {
+        dirtyDataConsumer.shutdown();
         if (!queue.isEmpty() && checkConsumer()) {
             consumer.consume();
         }
@@ -114,7 +127,7 @@ public void close() throws Exception {
     }
 
     /**
-     * 收集脏数据放入队列缓存中，记录放入失败的数目和存入队列中的总数目，如果放入失败的数目超过一定比列，那么manager任务失败
+     * 收集脏数据放入队列缓存中，记录放入失败的数目和存入队列中的总数目，如果放入失败的数目超过一定比例，那么manager任务失败
      */
     public void collectDirtyData(String dataInfo, String cause, String field) {
         DirtyDataEntity dirtyDataEntity = new DirtyDataEntity(dataInfo, System.currentTimeMillis(), cause, field);
@@ -124,6 +137,7 @@ public void collectDirtyData(String dataInfo, String cause, String field) {
         } catch (Exception ignored) {
             LOG.warn("dirty Data insert error ... Failed number: " + errorCount.incrementAndGet());
             LOG.warn("error dirty data:" + dirtyDataEntity.toString());
+            // TODO 这个失败比例可以作出调整
             if (errorCount.get() > Math.ceil(count.longValue() * 0.8)) {
                 throw new RuntimeException("The number of failed number reaches the limit, manager fails");
             }
diff --git a/core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/TestMain.java b/core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/TestMain.java
@@ -9,15 +9,25 @@
  * Date 2020/8/28 星期五
  */
 public class TestMain {
+    private static final Integer DATA_NUMBER = 10000;
+
     public static void main(String[] args) throws Exception {
-        Map<String, String> properties = new HashMap<>();
-        properties.put("type", "console");
-        properties.put("pluginPath", "/Users/wtz/IdeaProjects/flinkStreamSQL/sqlplugins");
+        Map<String, String> properties = new HashMap<>(8);
+        properties.put("type", "mysql");
+        properties.put("pluginPath", "/Users/wtz/IdeaProjects/flinkStreamSQLTemp/sqlplugins");
+        properties.put("url", "jdbc:mysql://kerberos01:3306/tiezhu");
+        properties.put("userName", "dtstack");
+        properties.put("password", "abc123");
+        properties.put("isCreatedTable", "false");
+        properties.put("batchSize", "1");
+        properties.put("tableName", "DirtyDataFromMysql_2020_09_14_10_51_50");
 
         DirtyDataManager manager = DirtyDataManager.newInstance(properties);
-        for (int i = 0; i < 100; i++) {
-            Thread.sleep(1000);
-            manager.collectDirtyData("testDirtyData", new Exception("testException").getMessage(), "testField");
+        for (int i = 0; i < DATA_NUMBER; i++) {
+            Thread.sleep(100);
+            manager.collectDirtyData("testDirtyData" + i,
+                    new Exception("testException" + i).getMessage(), "testField");
         }
+        manager.close();
     }
 }
diff --git a/dirtyDataManager/console/pom.xml b/dirtyDataManager/console/pom.xml
@@ -58,7 +58,7 @@
                         </goals>
                         <configuration>
                             <tasks>
-                                <copy todir="${basedir}/../../sqlplugins/dirtyData/print">
+                                <copy todir="${basedir}/../../sqlplugins/dirtyData/console">
                                     <fileset dir="target/">
                                         <include name="${project.artifactId}-${project.version}.jar"/>
                                     </fileset>
diff --git a/dirtyDataManager/mysql/src/main/java/com/dtstack/flink/sql/dirty/mysql/MysqlDirtyDataConsumer.java b/dirtyDataManager/mysql/src/main/java/com/dtstack/flink/sql/dirty/mysql/MysqlDirtyDataConsumer.java
@@ -19,90 +19,127 @@
 package com.dtstack.flink.sql.dirty.mysql;
 
 import com.dtstack.flink.sql.dirtyManager.consumer.AbstractDirtyDataConsumer;
+import com.dtstack.flink.sql.dirtyManager.entity.DirtyDataEntity;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * @author tiezhu
  * Company dtstack
  * Date 2020/8/27 星期四
  */
 public class MysqlDirtyDataConsumer extends AbstractDirtyDataConsumer {
-    //TODO 添加batchSize 和 定时任务
+
     private static final long serialVersionUID = -2959753658786001679L;
 
     private static final String DRIVER_NAME = "com.mysql.jdbc.Driver";
 
-    private final Object LOCK_STR = new Object();
+    private static final int CONN_VALID_TIME = 1000;
 
-    private boolean isCreatedTable = false;
+    private static final Integer FIELD_NUMBER = 5;
 
-    private final String[] tableField = {"id", "dirtyData", "processTime", "cause", "field"};
+    private final Object LOCK_STR = new Object();
 
-    private String SQL = "INSERT INTO ? (?, ?, ?, ?) VALUES (?, ?, ?, ?) ";
+    private final String[] tableField = {"id", "dirtyData", "processTime", "cause", "field"};
 
     private PreparedStatement statement;
 
     private Connection connection;
 
-    private String tableName;
+    private Long batchSize;
 
-    private void setStatement(String url,
-                              String userName,
-                              String password) throws ClassNotFoundException, SQLException {
+    private void beforeConsume(String url,
+                               String userName,
+                               String password,
+                               String tableName,
+                               boolean isCreatedTable) throws ClassNotFoundException, SQLException {
         synchronized (LOCK_STR) {
             Class.forName(DRIVER_NAME);
-
             connection = DriverManager.getConnection(url, userName, password);
-            statement = connection.prepareStatement(SQL);
+
+            // create table for dirty data
+            if (!isCreatedTable) {
+                createTable(tableName);
+            }
+
+            String insertField = Arrays.stream(tableField)
+                    .map(this::quoteIdentifier)
+                    .collect(Collectors.joining(", "));
+            String insertSql = "INSERT INTO " + quoteIdentifier(tableName)
+                    + "(" + insertField + ") VALUES (?, ?, ?, ?, ?)";
+            statement = connection.prepareStatement(insertSql);
         }
     }
 
     private String quoteIdentifier(String tableName) {
-        return "\"" + tableName + "\"";
+        return "`" + tableName + "`";
     }
 
     /**
      * 创建存储脏数据的表
      *
      * @param tableName 表名
-     * @return 是否创建成功
      * @throws SQLException SQL异常
      */
-    private boolean createTable(String tableName) {
+    private void createTable(String tableName) throws SQLException {
+        Statement statement = null;
         try {
-            String defaultTable = "";
             String sql =
-                    "CREATE TABLE ` " + tableName + "` (" +
+                    "CREATE TABLE  IF NOT EXISTS  \n"
+                            + quoteIdentifier(tableName) + " (\n" +
                             "  `id` int(11) not null AUTO_INCREMENT,\n" +
-                            "  `dirtyData` varchar(100) DEFAULT NULL,\n" +
-                            "  `processTime` varchar(100) DEFAULT NULL,\n" +
-                            "  `cause` date DEFAULT NULL,\n" +
-                            "  `field` varchar(100) DEFAULT NULL,\n" +
+                            "  `dirtyData` varchar(255) DEFAULT NULL,\n" +
+                            "  `processTime` varchar(255) DEFAULT NULL,\n" +
+                            "  `cause` varchar(255) DEFAULT NULL,\n" +
+                            "  `field` varchar(255) DEFAULT NULL,\n" +
                             "  PRIMARY KEY (id)\n" +
                             ") DEFAULT CHARSET=utf8;";
-            return statement.execute(sql);
+            statement = connection.createStatement();
+            statement.execute(sql);
         } catch (SQLException e) {
             throw new RuntimeException("create table error !", e);
+        } finally {
+            if (statement != null && !statement.isClosed()) {
+                statement.close();
+            }
         }
     }
 
     @Override
     public void consume() throws Exception {
-        if (!isCreatedTable) {
-            createTable(tableName);
+        DirtyDataEntity entity = queue.take();
+        count++;
+        List<String> data = new ArrayList<>();
+        data.add(String.valueOf(count));
+        Collections.addAll(data, entity.get());
+        for (int i = 0; i < FIELD_NUMBER; i++) {
+            statement.setString(i + 1, data.get(i));
         }
 
+        statement.addBatch();
+
+        if (count % batchSize == 0) {
+            statement.executeBatch();
+        }
     }
 
     @Override
     public void close() {
+        isRunning.compareAndSet(true, false);
+
         try {
-            if (connection != null && !connection.isValid(1000)) {
+            if (connection != null && !connection.isValid(CONN_VALID_TIME)) {
                 connection.close();
             }
 
@@ -116,11 +153,18 @@ public void close() {
 
     @Override
     public void init(Map<String, String> properties) throws Exception {
-        tableName = properties.get("tableName");
+        SimpleDateFormat timeFormat = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss");
+        String tableName = properties.getOrDefault("tableName",
+                "DirtyDataFromMysql_" + timeFormat.format(System.currentTimeMillis()));
         String userName = properties.get("userName");
         String password = properties.get("password");
         String url = properties.get("url");
-        isCreatedTable = Boolean.parseBoolean(properties.get("isCreatedTable"));
-        setStatement(url, userName, password);
+        batchSize = Long.parseLong(properties.getOrDefault("batchSize", "10000"));
+        errorLimit = Long.parseLong(properties.getOrDefault("errorLimit", "1000"));
+
+        boolean isCreatedTable = Boolean.parseBoolean(
+                properties.getOrDefault("isCreatedTable", "false"));
+
+        beforeConsume(url, userName, password, tableName, isCreatedTable);
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -94,4 +94,12 @@ public String toString() {`
`94`	`94`	`", field='" + field + '\'' +`
`95`	`95`	`'}';`
`96`	`96`	`}`
	`97`	`+`
	`98`	`+ /**`
	`99`	`+ * 获取脏数据信息，返回字符数组`
	`100`	`+ * @return 脏数据信息字符数组`
	`101`	`+ */`
	`102`	`+ public String[] get() {`
	`103`	`+ return new String[] {dirtyData, String.valueOf(processDate), cause, field};`
	`104`	`+ }`
`97`	`105`	`}`