[opt] code opt

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/dirtyManager/consumer/AbstractDirtyDataConsumer.java

@@ -26,6 +26,7 @@
 import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * @author tiezhu
@@ -38,9 +39,9 @@ public abstract class AbstractDirtyDataConsumer implements Runnable, Serializabl
     protected static final Logger LOG = LoggerFactory.getLogger(AbstractDirtyDataConsumer.class);
 
     protected Long errorLimit = 1000L;
-    protected Long errorCount = 0L;
+    protected AtomicLong errorCount = new AtomicLong(0L);
 
-    protected long count = 0;
+    protected AtomicLong count = new AtomicLong(0L);
 
     public AtomicBoolean isRunning = new AtomicBoolean(true);
 
@@ -82,9 +83,9 @@ public void run() {
             }
         } catch (Exception e) {
             LOG.error("consume dirtyData error", e);
-            errorCount++;
-            if (errorCount.equals(errorLimit)) {
-                throw new RuntimeException("脏数据消费失败达到上限，任务失败");
+            errorCount.incrementAndGet();
+            if (errorCount.get() == errorLimit) {
+                throw new RuntimeException("The task failed due to the number of dirty data consume failed reached the limit " + errorLimit);
             }
         }
     }

core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/DirtyDataManager.java

@@ -55,7 +55,7 @@ public class DirtyDataManager implements Serializable {
     /**
      * 写入队列阻塞时间
      */
-    private long blockingInterval = 60;
+    private long blockingInterval;
 
     /**
      * 缓存脏数据信息队列
@@ -72,6 +72,8 @@ public class DirtyDataManager implements Serializable {
      */
     private final AtomicLong errorCount = new AtomicLong(0);
 
+    private double errorLimitRate;
+
     public static AbstractDirtyDataConsumer consumer;
 
     private static ThreadPoolExecutor dirtyDataConsumer;
@@ -82,12 +84,17 @@ public class DirtyDataManager implements Serializable {
 
     private final static String DEFAULT_TYPE = "console";
 
+    private final static String DEFAULT_ERROR_LIMIT_RATE = "0.8";
+
+    private final static String DEFAULT_BLOCKING_INTERVAL = "60";
+
     /**
      * 通过参数生成manager实例，并同时将consumer实例化
      */
     public static DirtyDataManager newInstance(Map<String, String> properties) throws Exception {
         DirtyDataManager manager = new DirtyDataManager();
-        manager.blockingInterval = Long.parseLong(properties.getOrDefault("blockingInterval", "60"));
+        manager.blockingInterval = Long.parseLong(properties.getOrDefault("blockingInterval", DEFAULT_BLOCKING_INTERVAL));
+        manager.errorLimitRate = Double.parseDouble(properties.getOrDefault("errorLimitRate", DEFAULT_ERROR_LIMIT_RATE));
         consumer = createConsumer(properties);
         consumer.init(properties);
         consumer.setQueue(manager.queue);
@@ -119,11 +126,10 @@ private static AbstractDirtyDataConsumer createConsumer(Map<String, String> prop
      */
     public void close() throws Exception {
         dirtyDataConsumer.shutdown();
-        if (!queue.isEmpty() && checkConsumer()) {
-            consumer.consume();
+        if (checkConsumer()) {
+            LOG.info("dirty consumer is closing ...");
+            consumer.close();
         }
-        LOG.info("dirty consumer is closing ...");
-        consumer.close();
     }
 
     /**
@@ -137,8 +143,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)) {
+            if (errorCount.get() > Math.ceil(count.longValue() * errorLimitRate)) {
                 throw new RuntimeException("The number of failed number reaches the limit, manager fails");
             }
         }

core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/TestMain.java

@@ -9,7 +9,7 @@
  * Date 2020/8/28 星期五
  */
 public class TestMain {
-    private static final Integer DATA_NUMBER = 10000;
+    private static final Integer DATA_NUMBER = 1000;
 
     public static void main(String[] args) throws Exception {
         Map<String, String> properties = new HashMap<>(8);
@@ -27,7 +27,9 @@ public static void main(String[] args) throws Exception {
             Thread.sleep(100);
             manager.collectDirtyData("testDirtyData" + i,
                     new Exception("testException" + i).getMessage(), "testField");
+            if (i == 500) {
+                manager.close();
+            }
         }
-        manager.close();
     }
 }

dirtyDataManager/console/src/main/java/com/dtstack/flink/sql/dirty/print/ConsoleDirtyDataConsumer.java

@@ -38,7 +38,7 @@ public class ConsoleDirtyDataConsumer extends AbstractDirtyDataConsumer {
     @Override
     public void consume() throws InterruptedException {
         DirtyDataEntity dataEntity = queue.take();
-        count++;
+        count.incrementAndGet();
         LOG.warn("get dirtyData: " + dataEntity.getDirtyData() + "\n"
                 + "cause: " + dataEntity.getCause() + "\n"
                 + "processTime: " + dataEntity.getProcessDate() + "\n"

dirtyDataManager/mysql/src/main/java/com/dtstack/flink/sql/dirty/mysql/MysqlDirtyDataConsumer.java

@@ -119,17 +119,18 @@ private void createTable(String tableName) throws SQLException {
     @Override
     public void consume() throws Exception {
         DirtyDataEntity entity = queue.take();
-        count++;
+        count.incrementAndGet();
+
         List<String> data = new ArrayList<>();
-        data.add(String.valueOf(count));
+        data.add(String.valueOf(count.get()));
         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) {
+        if (count.get() % batchSize == 0) {
             statement.executeBatch();
         }
     }