代码整理

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java

@@ -2,20 +2,19 @@ package com.alibaba.otter.canal.adapter.launcher.loader;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
-import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
-import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
-import com.alibaba.otter.canal.client.adapter.support.Dml;
-import com.alibaba.otter.canal.protocol.FlatMessage;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
+import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
 import com.alibaba.otter.canal.client.adapter.OuterAdapter;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
 import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
+import com.alibaba.otter.canal.protocol.FlatMessage;
 import com.alibaba.otter.canal.protocol.Message;
 
 /**
@@ -44,21 +43,21 @@ public abstract class AbstractCanalAdapterWorker {
     protected void writeOut(final Message message) {
         List<Future<Boolean>> futures = new ArrayList<>();
         // 组间适配器并行运行
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
+        canalOuterAdapters.forEach(outerAdapters -> {
             final List<OuterAdapter> adapters = outerAdapters;
             futures.add(groupInnerExecutorService.submit(() -> {
                 try {
                     // 组内适配器穿行运行，尽量不要配置组内适配器
-                    for (final OuterAdapter c : adapters) {
+                    adapters.forEach(adapter -> {
                         long begin = System.currentTimeMillis();
-                        MessageUtil.parse4Dml(canalDestination, message, c::sync);
+                        MessageUtil.parse4Dml(canalDestination, message, adapter::sync);
 
                         if (logger.isDebugEnabled()) {
                             logger.debug("{} elapsed time: {}",
-                                c.getClass().getName(),
+                                adapter.getClass().getName(),
                                 (System.currentTimeMillis() - begin));
                         }
-                    }
+                    });
                     return true;
                 } catch (Exception e) {
                     return false;
@@ -67,79 +66,94 @@ public abstract class AbstractCanalAdapterWorker {
 
             // 等待所有适配器写入完成
             // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            for (Future<Boolean> f : futures) {
+            futures.forEach(future -> {
                 try {
-                    if (!f.get()) {
+                    if (!future.get()) {
                         logger.error("Outer adapter write failed");
                     }
                 } catch (InterruptedException | ExecutionException e) {
                     // ignore
                 }
-            }
-        }
+            });
+        });
     }
 
     protected void writeOut(final FlatMessage flatMessage) {
         List<Future<Boolean>> futures = new ArrayList<>();
         // 组间适配器并行运行
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-            final List<OuterAdapter> adapters = outerAdapters;
-            futures.add(groupInnerExecutorService.submit(new Callable<Boolean>() {
-
-                @Override
-                public Boolean call() {
-                    try {
-                        // 组内适配器穿行运行，尽量不要配置组内适配器
-                        for (OuterAdapter c : adapters) {
-                            long begin = System.currentTimeMillis();
-                            Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage);
-                            c.sync(dml);
-                            if (logger.isDebugEnabled()) {
-                                logger.debug("{} elapsed time: {}",
-                                    c.getClass().getName(),
-                                    (System.currentTimeMillis() - begin));
-                            }
+        canalOuterAdapters.forEach(outerAdapters -> {
+            futures.add(groupInnerExecutorService.submit(() -> {
+                try {
+                    // 组内适配器穿行运行，尽量不要配置组内适配器
+                    outerAdapters.forEach(adapter -> {
+                        long begin = System.currentTimeMillis();
+                        Dml dml = MessageUtil.flatMessage2Dml(canalDestination, flatMessage);
+                        adapter.sync(dml);
+                        if (logger.isDebugEnabled()) {
+                            logger.debug("{} elapsed time: {}",
+                                adapter.getClass().getName(),
+                                (System.currentTimeMillis() - begin));
                         }
-                        return true;
-                    } catch (Exception e) {
-                        return false;
-                    }
+                    });
+                    return true;
+                } catch (Exception e) {
+                    return false;
                 }
             }));
 
             // 等待所有适配器写入完成
             // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
-            for (Future<Boolean> f : futures) {
+            futures.forEach(future -> {
                 try {
-                    if (!f.get()) {
+                    if (!future.get()) {
                         logger.error("Outer adapter write failed");
                     }
                 } catch (InterruptedException | ExecutionException e) {
                     // ignore
                 }
-            }
+            });
+        });
+    }
+
+    public void start() {
+        if (!running) {
+            thread = new Thread(this::process);
+            thread.setUncaughtExceptionHandler(handler);
+            thread.start();
+            running = true;
         }
     }
 
-    protected void stopOutAdapters() {
-        if (thread != null) {
-            try {
-                thread.join();
-            } catch (InterruptedException e) {
-                // ignore
+    protected abstract void process();
+
+    public void stop() {
+        try {
+            if (!running) {
+                return;
             }
-        }
-        groupInnerExecutorService.shutdown();
-        logger.info("topic connectors' worker thread dead!");
-        for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-            for (OuterAdapter adapter : outerAdapters) {
-                adapter.destroy();
+
+            closeConnection();
+            running = false;
+
+            syncSwitch.release(canalDestination);
+
+            logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination);
+            if (thread != null) {
+                try {
+                    thread.join();
+                } catch (InterruptedException e) {
+                    // ignore
+                }
             }
+            groupInnerExecutorService.shutdown();
+            logger.info("destination {} adapters worker thread dead!", canalDestination);
+            canalOuterAdapters.forEach(outerAdapters -> outerAdapters.forEach(OuterAdapter::destroy));
+            logger.info("destination {} all adapters destroyed!", canalDestination);
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
         }
-        logger.info("topic all connectors destroyed!");
     }
 
-    public abstract void start();
+    protected abstract void closeConnection();
 
-    public abstract void stop();
 }

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java

@@ -37,59 +37,18 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
         this.flatMessage = flatMessage;
         connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage);
         // connector.setSessionTimeout(1L, TimeUnit.MINUTES);
-
-        // super.initSwitcher(topic);
     }
 
     @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(() -> process());
-            thread.setUncaughtExceptionHandler(handler);
-            running = true;
-            thread.start();
-        }
+    protected  void closeConnection(){
+        connector.stopRunning();
     }
 
     @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-
-            connector.stopRunning();
-            running = false;
-
-            // if (switcher != null && !switcher.state()) {
-            // switcher.set(true);
-            // }
-
-            if (thread != null) {
-                try {
-                    thread.join();
-                } catch (InterruptedException e) {
-                    // ignore
-                }
-            }
-            groupInnerExecutorService.shutdown();
-            logger.info("topic {} connectors' worker thread dead!", this.topic);
-            for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-                for (OuterAdapter adapter : outerAdapters) {
-                    adapter.destroy();
-                }
-            }
-            logger.info("topic {} all connectors destroyed!", this.topic);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
+    protected void process() {
         while (!running)
             ;
         ExecutorService executor = Executors.newSingleThreadExecutor();
-        // final AtomicBoolean executing = new AtomicBoolean(true);
         while (running) {
             try {
                 logger.info("=============> Start to connect topic: {} <=============", this.topic);
@@ -99,7 +58,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
                 logger.info("=============> Subscribe topic: {} succeed <=============", this.topic);
                 while (running) {
                     try {
-                        // switcher.get(); //等待开关开启
+                        syncSwitch.get(canalDestination);
 
                         List<?> messages;
                         if (!flatMessage) {

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java

@@ -135,20 +135,18 @@ public class CanalAdapterLoader {
     public void destroy() {
         if (canalWorkers.size() > 0) {
             ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size());
-            for (CanalAdapterWorker v : canalWorkers.values()) {
-                final CanalAdapterWorker caw = v;
-                stopExecutorService.submit(caw::stop);
+            for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) {
+                stopExecutorService.submit(canalAdapterWorker::stop);
             }
             stopExecutorService.shutdown();
         }
 
         if (canalMQWorker.size() > 0) {
-            ExecutorService stopMQWokerService = Executors.newFixedThreadPool(canalMQWorker.size());
-            for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) {
-                final AbstractCanalAdapterWorker worker = tmp;
-                stopMQWokerService.submit(worker::stop);
+            ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size());
+            for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) {
+                stopMQWorkerService.submit(canalAdapterMQWorker::stop);
             }
-            stopMQWokerService.shutdown();
+            stopMQWorkerService.shutdown();
         }
         logger.info("All canal adapters destroyed");
     }

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java

@@ -36,38 +36,12 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
     }
 
     @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(new Runnable() {
-
-                @Override
-                public void run() {
-                    process();
-                }
-            });
-            thread.setUncaughtExceptionHandler(handler);
-            running = true;
-            thread.start();
-        }
+    protected void closeConnection() {
+        connector.stopRunning();
     }
 
     @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-            connector.stopRunning();
-            running = false;
-            logger.info("Stop topic {} out adapters begin", this.topic);
-            stopOutAdapters();
-            logger.info("Stop topic {} out adapters end", this.topic);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
+    protected void process() {
         while (!running)
             ;
         ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -84,25 +58,27 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
 
                         final Message message = connector.getWithoutAck(1);
                         if (message != null) {
-                            executor.submit(new Runnable() {
-
-                                @Override
-                                public void run() {
-                                    try {
-                                        if (logger.isDebugEnabled()) {
-                                            logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size());
-                                        }
-                                        long begin = System.currentTimeMillis();
-                                        writeOut(message);
-                                        long now = System.currentTimeMillis();
-                                        if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
-                                            logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
-                                        }
-                                    } catch (Exception e) {
-                                        logger.error(e.getMessage(), e);
+                            executor.submit(() -> {
+                                try {
+                                    if (logger.isDebugEnabled()) {
+                                        logger.debug("topic: {} batchId: {} batchSize: {} ",
+                                            topic,
+                                            message.getId(),
+                                            message.getEntries().size());
+                                    }
+                                    long begin = System.currentTimeMillis();
+                                    writeOut(message);
+                                    long now = System.currentTimeMillis();
+                                    if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
+                                        logger.error("topic: {} batchId {} elapsed time: {} ms",
+                                            topic,
+                                            message.getId(),
+                                            now - begin);
                                     }
-                                    connector.ack(message.getId());
+                                } catch (Exception e) {
+                                    logger.error(e.getMessage(), e);
                                 }
+                                connector.ack(message.getId());
                             });
                         } else {
                             logger.debug("Message is null");

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java

@@ -57,49 +57,12 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
     }
 
     @Override
-    public void start() {
-        if (!running) {
-            thread = new Thread(this::process);
-            thread.setUncaughtExceptionHandler(handler);
-            thread.start();
-            running = true;
-        }
+    protected void closeConnection() {
+        connector.stopRunning();
     }
 
     @Override
-    public void stop() {
-        try {
-            if (!running) {
-                return;
-            }
-
-            connector.stopRunning();
-            running = false;
-
-            syncSwitch.release(canalDestination);
-
-            logger.info("destination {} is waiting for adapters' worker thread die!", canalDestination);
-            if (thread != null) {
-                try {
-                    thread.join();
-                } catch (InterruptedException e) {
-                    // ignore
-                }
-            }
-            groupInnerExecutorService.shutdown();
-            logger.info("destination {} adapters' worker thread dead!", canalDestination);
-            for (List<OuterAdapter> outerAdapters : canalOuterAdapters) {
-                for (OuterAdapter adapter : outerAdapters) {
-                    adapter.destroy();
-                }
-            }
-            logger.info("destination {} all adapters destroyed!", canalDestination);
-        } catch (Exception e) {
-            logger.error(e.getMessage(), e);
-        }
-    }
-
-    private void process() {
+    protected void process() {
         while (!running)
             ; // waiting until running == true
         while (running) {
@@ -129,11 +92,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
                         int size = message.getEntries().size();
 
                         if (batchId == -1 || size == 0) {
-                            try {
-                                Thread.sleep(1000);
-                            } catch (InterruptedException e) {
-                                // ignore
-                            }
+                            Thread.sleep(1000);
                         } else {
                             if (logger.isDebugEnabled()) {
                                 logger.debug("destination: {} batchId: {} batchSize: {} ",
@@ -143,18 +102,18 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
                             }
                             long begin = System.currentTimeMillis();
                             writeOut(message);
-                            long now = System.currentTimeMillis();
                             if (logger.isDebugEnabled()) {
                                 logger.debug("destination: {} batchId: {} elapsed time: {} ms",
                                     this.canalDestination,
                                     batchId,
-                                    now - begin);
+                                    System.currentTimeMillis() - begin);
                             }
                         }
                         connector.ack(batchId); // 提交确认
                     } catch (Exception e) {
                         connector.rollback(batchId); // 处理失败, 回滚数据
-                        throw e;
+                        logger.error("sync error!", e);
+                        Thread.sleep(500);
                     }
                 }