新增失败回调

core/src/main/java/com/tbyf/cdcengine2/core/AbstractCdcEngine.java

@@ -41,7 +41,9 @@ public abstract class AbstractCdcEngine<T extends AbstractCdcEngine<T>> {
 
     private volatile boolean running = true;
 
-    private volatile List<ChangeListener> changeListeners = new ArrayList<>();
+    private ChangeHandler changeHandler;
+
+    private Runnable errorHandler;
 
     public AbstractCdcEngine() {
         debeziumProps.setProperty(Constants.DATABASE_SERVER_NAME_PROP, Constants.DATABASE_SERVER_NAME_VALUE);
@@ -49,42 +51,47 @@ public abstract class AbstractCdcEngine<T extends AbstractCdcEngine<T>> {
         debeziumProps.setProperty("name", getClass().getSimpleName());
     }
 
-    public void addListener(ChangeListener listener) {
-        ArrayList<ChangeListener> changeListeners = new ArrayList<>(this.changeListeners);
-        changeListeners.add(listener);
-        this.changeListeners = changeListeners;
+    @SuppressWarnings("unchecked")
+    private T self() {
+        return (T) this;
+    }
+
+    public T onChange(ChangeHandler changeHandler) {
+        if (changeHandler == null) {
+            throw new IllegalArgumentException("changeHandler must not be null");
+        }
+        this.changeHandler = changeHandler;
+        return self();
+    }
+
+    public T onError(Runnable errorHandler) {
+        this.errorHandler = errorHandler;
+        return self();
     }
 
     public void start() {
         synchronized (startupStopMonitor) {
             logger.info("开始启动CDC引擎...");
+            if (this.changeHandler == null) {
+                throw new IllegalStateException("changeHandler not set");
+            }
             this.running = true;
             consumerThread = new Thread(() -> {
-                List<ChangeListener> changeListeners = null;
                 while (running) {
-                    changeListeners = this.changeListeners;
                     try {
                         List<ChangedRecord> buffer = new ArrayList<>();
                         recordQueue.drainTo(buffer);
                         if (buffer.isEmpty()) {
                             ChangedRecord record = recordQueue.take();
                             try {
-                                for (ChangeListener listener : changeListeners) {
-                                    if (listener.supports(record.schema(), record.table())) {
-                                        listener.onChange(record);
-                                    }
-                                }
+                                this.changeHandler.handle(record);
                             } catch (Exception e) {
                                 logger.error("处理{}时发生了异常", record, e);
                             }
                         } else {
                             for (ChangedRecord record : buffer) {
                                 try {
-                                    for (ChangeListener listener : changeListeners) {
-                                        if (listener.supports(record.schema(), record.table())) {
-                                            listener.onChange(record);
-                                        }
-                                    }
+                                    changeHandler.handle(record);
                                 } catch (Exception e) {
                                     logger.error("处理{}时发生了异常", record, e);
                                 }
@@ -98,14 +105,9 @@ public abstract class AbstractCdcEngine<T extends AbstractCdcEngine<T>> {
                 List<ChangedRecord> buffer = new ArrayList<>();
                 recordQueue.drainTo(buffer);
                 if (!buffer.isEmpty()) {
-                    changeListeners = this.changeListeners;
                     for (ChangedRecord record : buffer) {
                         try {
-                            for (ChangeListener listener : changeListeners) {
-                                if (listener.supports(record.schema(), record.table())) {
-                                    listener.onChange(record);
-                                }
-                            }
+                            changeHandler.handle(record);
                         } catch (Exception e) {
                             logger.error("处理{}时发生了异常", record, e);
                         }
@@ -166,6 +168,11 @@ public abstract class AbstractCdcEngine<T extends AbstractCdcEngine<T>> {
                 consumerThread.interrupt();
             }
             this.consumerThread = null;
+
+            Runnable errorHandler = this.errorHandler;
+            if (errorHandler != null) {
+                errorHandler.run();
+            }
         }
     }
 

core/src/main/java/com/tbyf/cdcengine2/core/ChangeHandler.java

+package com.tbyf.cdcengine2.core;
+
+public interface ChangeHandler {
+
+    void handle(ChangedRecord record);
+}