[feat] 完成 print dirty consumer

FlechazoW · FlechazoW · commit 055bcf5824f1 · 2020-09-03T21:49:10.000+08:00
diff --git a/core/pom.xml b/core/pom.xml
@@ -123,6 +123,17 @@
             <artifactId>joda-time</artifactId>
             <version>2.5</version>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <version>1.7.30</version>
+        </dependency>
+
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <version>1.2.17</version>
+        </dependency>
     </dependencies>
 
     <build>
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
@@ -19,26 +19,37 @@
 package com.dtstack.flink.sql.dirtyManager.consumer;
 
 import com.dtstack.flink.sql.dirtyManager.entity.DirtyDataEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
+import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * @author tiezhu
  * Company dtstack
  * Date 2020/8/27 星期四
  */
-public abstract class AbstractDirtyDataConsumer extends Thread implements  Serializable {
-    private static final long serialVersionUID = -6058598201315176687L;
+public abstract class AbstractDirtyDataConsumer implements Runnable, Serializable {
+    protected static final long serialVersionUID = -6058598201315176687L;
 
-    private long count = 0;
+    protected static final Logger LOG = LoggerFactory.getLogger(AbstractDirtyDataConsumer.class);
+
+    protected long errorLimit = 1000;
+    protected long errorCount = 0;
+
+    protected long count = 0;
+
+    public AtomicBoolean isRunning = new AtomicBoolean(true);
 
     protected LinkedBlockingQueue<DirtyDataEntity> queue;
 
     /**
      * 消费队列数据
      */
-    public abstract void consume() throws InterruptedException;
+    public abstract void consume() throws Exception;
 
     /**
      * 关闭消费者，需要释放资源
@@ -48,16 +59,26 @@ public abstract class AbstractDirtyDataConsumer extends Thread implements  Seria
     /**
      * 初始化消费者，初始化定时任务
      */
-    public void init() {
-
-    }
+    public abstract void init(Map<String, String> properties);
 
     @Override
     public void run() {
-
+        try {
+            LOG.info("start to consume dirty data");
+            while (isRunning.get()) {
+                consume();
+            }
+        } catch (Exception e) {
+            LOG.error("consume dirtyData error");
+            errorCount++;
+            if (errorCount == errorLimit) {
+                throw new RuntimeException("脏数据消费失败达到上限，任务失败");
+            }
+        }
     }
 
-    public void setQueue(LinkedBlockingQueue<DirtyDataEntity> queue) {
+    public AbstractDirtyDataConsumer setQueue(LinkedBlockingQueue<DirtyDataEntity> queue) {
         this.queue = queue;
+        return this;
     }
 }
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
@@ -22,13 +22,13 @@
 import com.dtstack.flink.sql.dirtyManager.consumer.AbstractDirtyDataConsumer;
 import com.dtstack.flink.sql.dirtyManager.entity.DirtyDataEntity;
 import com.dtstack.flink.sql.util.PluginUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.Serializable;
 import java.lang.reflect.Constructor;
-import java.net.URL;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -41,6 +41,8 @@
 public class DirtyDataManager implements Serializable {
     private static final long serialVersionUID = 7190970299538893497L;
 
+    private static final Logger LOG = LoggerFactory.getLogger(DirtyDataManager.class);
+
     private static final String CLASS_PRE_STR = "com.dtstack.flink.sql.dirty";
 
     private static final String CLASS_POST_STR = "DirtyDataConsumer";
@@ -55,53 +57,72 @@ public class DirtyDataManager implements Serializable {
     /**
      * 缓存脏数据信息队列
      */
-    private LinkedBlockingQueue<DirtyDataEntity> queue = new LinkedBlockingQueue<>();
+    public final LinkedBlockingQueue<DirtyDataEntity> queue = new LinkedBlockingQueue<>();
 
     /**
      * 统计manager收集到的脏数据条数
      */
-    private AtomicLong count = new AtomicLong(0);
-
-    private AbstractDirtyDataConsumer consumer;
+    private final AtomicLong count = new AtomicLong(0);
 
-    public static DirtyDataManager newInstance() {
+    public static AbstractDirtyDataConsumer consumer;
 
-        return new DirtyDataManager();
+    /**
+     * 通过参数生成manager实例，并同时将consumer实例化
+     */
+    public static DirtyDataManager newInstance(Map<String, String> properties) throws Exception {
+        DirtyDataManager manager = new DirtyDataManager();
+        manager.blockingInterval = Long.parseLong(properties.getOrDefault("blockingInterval", "60"));
+        consumer = createConsumer(properties);
+        consumer.init(properties);
+        Thread dirtyDataConsumer = new Thread(consumer.setQueue(manager.queue), "dirtyData Consumer");
+        dirtyDataConsumer.start();
+        return manager;
     }
 
-    public AbstractDirtyDataConsumer createConsumer(Map<String, String> properties) throws Exception {
-        // 利用类加载的方式动态加载
+    /**
+     * 通过动态加载的方式加载Consumer
+     */
+    private static AbstractDirtyDataConsumer createConsumer(Map<String, String> properties) throws Exception {
         String type = properties.getOrDefault("type", "print");
         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);
 
         return ClassLoaderManager.newInstance(consumerJar, cl -> {
             Class<?> clazz = cl.loadClass(className);
-            Constructor<?> constructor = clazz.getConstructor(String.class);
-            return (AbstractDirtyDataConsumer) constructor.newInstance(type + CLASS_POST_STR);
+            Constructor<?> constructor = clazz.getConstructor();
+            return (AbstractDirtyDataConsumer) constructor.newInstance();
         });
     }
 
-    public void close() throws InterruptedException {
-        if (!queue.isEmpty()) {
-            flush();
+    /**
+     * 脏数据收集任务停止，任务停止之前，需要将队列中所有的数据清空
+     */
+    public void close() throws Exception {
+        if (!queue.isEmpty() && checkConsumer()) {
+            consumer.consume();
         }
-    }
-
-    public void flush() throws InterruptedException {
-        consumer.setQueue(queue);
-        consumer.consume();
+        LOG.info("dirty consumer is closing ......");
+        consumer.close();
     }
 
     // 收集脏数据
     public void collectDirtyData(String dataInfo, String cause, String field) throws InterruptedException {
         DirtyDataEntity dirtyDataEntity = new DirtyDataEntity(dataInfo, System.currentTimeMillis(), cause, field);
         queue.offer(dirtyDataEntity, blockingInterval, TimeUnit.MILLISECONDS);
         count.incrementAndGet();
-        consumer.setQueue(queue);
     }
 
+    /**
+     * 查看consumer当前状态
+     */
+    public boolean checkConsumer() {
+        return consumer.isRunning.get();
+    }
+
+    /**
+     * 首字母大写
+     */
     private static String upperCaseFirstChar(String str) {
         return str.substring(0, 1).toUpperCase() + str.substring(1);
     }
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
@@ -0,0 +1,23 @@
+package com.dtstack.flink.sql.dirtyManager.manager;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author tiezhu
+ * Company dtstack
+ * Date 2020/8/28 星期五
+ */
+public class TestMain {
+    public static void main(String[] args) throws Exception {
+        Map<String, String> properties = new HashMap<>();
+        properties.put("type", "print");
+        properties.put("pluginPath", "/Users/wtz/IdeaProjects/flinkStreamSQL/sqlplugins");
+
+        DirtyDataManager manager = DirtyDataManager.newInstance(properties);
+        for (int i = 0; i < 100; i++) {
+            Thread.sleep(1000);
+            manager.collectDirtyData("testDirtyData", new Exception("testException").getMessage(), "testField");
+        }
+    }
+}
diff --git a/core/src/test/java/com/dtstack/flink/sql/dirtyManager/TestDirtyDataManager.java b/core/src/test/java/com/dtstack/flink/sql/dirtyManager/TestDirtyDataManager.java
@@ -1,9 +1,6 @@
 package com.dtstack.flink.sql.dirtyManager;
 
-import com.dtstack.flink.sql.dirtyManager.consumer.AbstractDirtyDataConsumer;
 import com.dtstack.flink.sql.dirtyManager.manager.DirtyDataManager;
-import org.junit.Assert;
-import org.junit.Test;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -14,14 +11,5 @@
  * Date 2020/8/28 星期五
  */
 public class TestDirtyDataManager {
-    @Test
-    public void testCreateConsumer() throws Exception {
-        Map<String, String> properties = new HashMap<>();
-        properties.put("type", "print");
-        properties.put("pluginPath", "/Users/wtz/IdeaProjects/flinkStreamSQL/sqlplugins");
 
-        DirtyDataManager manager = DirtyDataManager.newInstance();
-       AbstractDirtyDataConsumer consumer = manager.createConsumer(properties);
-        Assert.assertNotNull(consumer);
-    }
 }
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,11 +19,14 @@
 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.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * @author tiezhu
@@ -45,6 +48,8 @@ public class MysqlDirtyDataConsumer extends AbstractDirtyDataConsumer {
 
     private Connection connection;
 
+    private String tableName;
+
     private PreparedStatement getStatement(String url,
                                            String userName,
                                            String password) throws ClassNotFoundException, SQLException {
@@ -79,12 +84,19 @@ private boolean createTable(String tableName) throws SQLException {
     }
 
     @Override
-    public void consume() {
-
+    public void consume() throws Exception{
+        if (!isCreatedTable) {
+            createTable("tableName");
+        }
     }
 
     @Override
     public void close() {
 
     }
+
+    @Override
+    public void init(Map<String, String> properties) {
+
+    }
 }
diff --git a/dirtyDataManager/print/pom.xml b/dirtyDataManager/print/pom.xml
@@ -73,5 +73,24 @@
             </plugin>
         </plugins>
     </build>
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.8.2</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <version>1.7.30</version>
+        </dependency>
+
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <version>1.2.17</version>
+        </dependency>
+    </dependencies>
 
 </project>
diff --git a/dirtyDataManager/print/src/main/java/com/dtstack/flink/sql/dirty/print/PrintDirtyDataConsumer.java b/dirtyDataManager/print/src/main/java/com/dtstack/flink/sql/dirty/print/PrintDirtyDataConsumer.java
@@ -23,6 +23,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+
 /**
  * @author tiezhu
  * Company dtstack
@@ -36,6 +38,7 @@ public class PrintDirtyDataConsumer extends AbstractDirtyDataConsumer {
     @Override
     public void consume() throws InterruptedException {
         DirtyDataEntity dataEntity = queue.take();
+        count++;
         LOG.warn("get dirtyData: " + dataEntity.getDirtyData() + "\n"
                 + "cause: " + dataEntity.getCause() + "\n"
                 + "processTime: " + dataEntity.getProcessDate() + "\n"
@@ -44,6 +47,11 @@ public void consume() throws InterruptedException {
 
     @Override
     public void close() {
-        // Do nothing
+        isRunning.compareAndSet(true, false);
+    }
+
+    @Override
+    public void init(Map<String, String> properties) {
+        LOG.info("Log dirty consumer init ......");
     }
 }
diff --git a/dirtyDataManager/print/src/test/java/com/dtstack/flink/sql/dirty/print/TestPrintDirtyDataConsumer.java b/dirtyDataManager/print/src/test/java/com/dtstack/flink/sql/dirty/print/TestPrintDirtyDataConsumer.java
@@ -0,0 +1,12 @@
+package com.dtstack.flink.sql.dirty.print;
+
+/**
+ * @author tiezhu
+ * Company dtstack
+ * Date 2020/8/28 星期五
+ */
+public class TestPrintDirtyDataConsumer {
+    public static void main(String[] args) {
+
+    }
+}
