修改adapter-launcher模块名为launcher

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

@@ -0,0 +1,56 @@
+package com.alibaba.otter.canal.adapter.launcher;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.annotation.Resource;
+
+import com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterLoader;
+import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+
+
+@SpringBootApplication
+public class CanalAdapterApplication {
+    private static final Logger logger = LoggerFactory.getLogger(CanalAdapterApplication.class);
+
+    private static CanalAdapterLoader adapterLoader;
+
+    public static void main(String[] args) {
+        new SpringApplicationBuilder(CanalAdapterApplication.class).run(args);
+    }
+
+    @Resource
+    private CanalClientConfig canalClientConf;
+
+    @PostConstruct
+    public void init() {
+        if (adapterLoader == null) {
+            try {
+                logger.info("## start the canal client adapters.");
+                adapterLoader = new CanalAdapterLoader(canalClientConf);
+                adapterLoader.init();
+                logger.info("## the canal client adapters are running now ......");
+            } catch (Throwable e) {
+                logger.error("## something goes wrong when starting up the canal client adapters:", e);
+                System.exit(0);
+            }
+        }
+    }
+
+    @PreDestroy
+    public void destroy() {
+        try {
+            logger.info("## stop the canal client adapters");
+            if (adapterLoader != null) {
+                adapterLoader.destroy();
+            }
+        } catch (Throwable e) {
+            logger.warn("## something goes wrong when stopping canal client adapters:", e);
+        } finally {
+            logger.info("## canal client adapters are down.");
+        }
+    }
+}

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java

@@ -0,0 +1,10 @@
+package com.alibaba.otter.canal.adapter.launcher.config;
+
+import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+@ConfigurationProperties(prefix = "canal.conf")
+@Component
+public class AdapterCanalConfig extends CanalClientConfig {
+}

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java

@@ -0,0 +1,35 @@
+package com.alibaba.otter.canal.adapter.launcher.config;
+
+import java.util.List;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+
+@Component
+@ConfigurationProperties(prefix = "adapter.conf.path")
+public class AdapterConfigPath {
+
+    private List<String> adapterConfigs;
+
+    public List<String> getAdapterConfigs() {
+        return adapterConfigs;
+    }
+
+    public void setAdapterConfigs(List<String> adapterConfigs) {
+        this.adapterConfigs = adapterConfigs;
+
+        if (adapterConfigs != null) {
+            AdapterConfigs.configs.clear();
+            for (String adapterConfig : adapterConfigs) {
+                int idx = adapterConfig.indexOf("/");
+                if (idx > -1) {
+                    String type = adapterConfig.substring(0, idx);
+                    String ymlFile = adapterConfig.substring(idx + 1);
+                    AdapterConfigs.configs.put(type, ymlFile);
+                }
+            }
+        }
+    }
+}

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

@@ -0,0 +1,160 @@
+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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * 适配器工作线程抽象类
+ *
+ * @author machengyuan 2018-8-19 下午11:30:49
+ * @version 1.0.0
+ */
+public abstract class AbstractCanalAdapterWorker {
+
+    protected final Logger                    logger  = LoggerFactory.getLogger(this.getClass());
+
+    protected String                          canalDestination;                                                 // canal实例
+    protected List<List<CanalOuterAdapter>>   canalOuterAdapters;                                               // 外部适配器
+    protected ExecutorService                 groupInnerExecutorService;                                        // 组内工作线程池
+    protected volatile boolean                running = false;                                                  // 是否运行中
+    protected Thread                          thread  = null;
+    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
+
+                                                          @Override
+                                                          public void uncaughtException(Thread t, Throwable e) {
+                                                              logger.error("parse events has an error", e);
+                                                          }
+                                                      };
+
+    protected void writeOut(final Message message) {
+        List<Future<Boolean>> futures = new ArrayList<>();
+        // 组间适配器并行运行
+        for (List<CanalOuterAdapter> outerAdapters : canalOuterAdapters) {
+            final List<CanalOuterAdapter> adapters = outerAdapters;
+            futures.add(groupInnerExecutorService.submit(new Callable<Boolean>() {
+
+                @Override
+                public Boolean call() {
+                    try {
+                        // 组内适配器穿行运行，尽量不要配置组内适配器
+                        for (final CanalOuterAdapter c : adapters) {
+                            long begin = System.currentTimeMillis();
+                            MessageUtil.parse4Dml(message, c::writeOut);
+
+                            if (logger.isDebugEnabled()) {
+                                logger.debug("{} elapsed time: {}",
+                                    c.getClass().getName(),
+                                    (System.currentTimeMillis() - begin));
+                            }
+                        }
+                        return true;
+                    } catch (Exception e) {
+                        return false;
+                    }
+                }
+            }));
+
+            // 等待所有适配器写入完成
+            // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
+            for (Future<Boolean> f : futures) {
+                try {
+                    if (!f.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<CanalOuterAdapter> outerAdapters : canalOuterAdapters) {
+    // final List<CanalOuterAdapter> adapters = outerAdapters;
+    // futures.add(groupInnerExecutorService.submit(new Callable<Boolean>() {
+    //
+    // @Override
+    // public Boolean call() {
+    // try {
+    // // 组内适配器穿行运行，尽量不要配置组内适配器
+    // for (CanalOuterAdapter c : adapters) {
+    // long begin = System.currentTimeMillis();
+    // Dml dml = MessageUtil.flatMessage2Dml(flatMessage);
+    // c.writeOut(dml);
+    // if (logger.isDebugEnabled()) {
+    // logger.debug("{} elapsed time: {}",
+    // c.getClass().getName(),
+    // (System.currentTimeMillis() - begin));
+    // }
+    // }
+    // return true;
+    // } catch (Exception e) {
+    // return false;
+    // }
+    // }
+    // }));
+    //
+    // // 等待所有适配器写入完成
+    // // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
+    // for (Future<Boolean> f : futures) {
+    // try {
+    // if (!f.get()) {
+    // logger.error("Outer adapter write failed");
+    // }
+    // } catch (InterruptedException | ExecutionException e) {
+    // // ignore
+    // }
+    // }
+    // }
+    // }
+
+    protected void writeOut(Message message, String topic) {
+        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);
+        }
+        if (logger.isDebugEnabled()) {
+            logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
+        }
+    }
+
+    protected void stopOutAdapters() {
+        if (thread != null) {
+            try {
+                thread.join();
+            } catch (InterruptedException e) {
+                // ignore
+            }
+        }
+        groupInnerExecutorService.shutdown();
+        logger.info("topic connectors' worker thread dead!");
+        for (List<CanalOuterAdapter> outerAdapters : canalOuterAdapters) {
+            for (CanalOuterAdapter adapter : outerAdapters) {
+                adapter.destroy();
+            }
+        }
+        logger.info("topic all connectors destroyed!");
+    }
+
+    public abstract void start();
+
+    public abstract void stop();
+}

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

@@ -0,0 +1,112 @@
+package com.alibaba.otter.canal.adapter.launcher.loader;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
+import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
+
+/**
+ * 外部适配器的加载器
+ *
+ * @version 1.0.0
+ */
+public class CanalAdapterLoader {
+
+    private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class);
+
+    private CanalClientConfig canalClientConfig;
+
+    private Map<String, CanalAdapterWorker> canalWorkers = new HashMap<>();
+
+    private ExtensionLoader<CanalOuterAdapter> loader;
+
+    public CanalAdapterLoader(CanalClientConfig canalClientConfig) {
+        this.canalClientConfig = canalClientConfig;
+    }
+
+    /**
+     * 初始化canal-client
+     */
+    public void init() {
+        // canal instances 和 mq topics 配置不能同时为空
+        // if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) {
+        //    throw new RuntimeException("Blank config property: canalInstances or canalMQTopics");
+        // }
+
+        loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class);
+
+        String canalServerHost = this.canalClientConfig.getCanalServerHost();
+        SocketAddress sa = null;
+        if (canalServerHost != null) {
+            String[] ipPort = canalServerHost.split(":");
+            sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1]));
+        }
+        String zkHosts = this.canalClientConfig.getZookeeperHosts();
+
+        // 初始化canal-client的适配器
+        if (canalClientConfig.getCanalInstances() != null) {
+            for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) {
+                List<List<CanalOuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
+
+                for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) {
+                    List<CanalOuterAdapter> canalOutConnectors = new ArrayList<>();
+                    for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) {
+                        loadConnector(c, canalOutConnectors);
+                    }
+                    canalOuterAdapterGroups.add(canalOutConnectors);
+                }
+                CanalAdapterWorker worker;
+                if (zkHosts != null) {
+                    worker = new CanalAdapterWorker(instance.getInstance(), zkHosts, canalOuterAdapterGroups);
+                } else {
+                    worker = new CanalAdapterWorker(instance.getInstance(), sa, canalOuterAdapterGroups);
+                }
+                canalWorkers.put(instance.getInstance(), worker);
+                worker.start();
+                logger.info("Start adapter for canal instance: {} succeed", instance.getInstance());
+            }
+        }
+    }
+
+    private void loadConnector(CanalOuterAdapterConfiguration config, List<CanalOuterAdapter> canalOutConnectors) {
+        try {
+            CanalOuterAdapter adapter = loader.getExtension(config.getName());
+            ClassLoader cl = Thread.currentThread().getContextClassLoader();
+            // 替换ClassLoader
+            Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader());
+            adapter.init(config);
+            Thread.currentThread().setContextClassLoader(cl);
+            canalOutConnectors.add(adapter);
+            logger.info("Load canal adapter: {} succeed", config.getName());
+        } catch (Exception e) {
+            logger.error("Load canal adapter: {} failed", config.getName(), e);
+        }
+    }
+
+    /**
+     * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁
+     */
+    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());
+            }
+            stopExecutorService.shutdown();
+        }
+        logger.info("All canal adapters destroyed");
+    }
+}

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

@@ -0,0 +1,186 @@
+package com.alibaba.otter.canal.adapter.launcher.loader;
+
+import java.net.SocketAddress;
+import java.util.List;
+import java.util.concurrent.Executors;
+
+import com.alibaba.otter.canal.client.CanalConnector;
+import com.alibaba.otter.canal.client.CanalConnectors;
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * 原生canal-server对应的client适配器工作线程
+ *
+ * @author machengyuan 2018-8-19 下午11:30:49
+ * @version 1.0.0
+ */
+public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
+
+    private static final int BATCH_SIZE = 50;
+    private static final int SO_TIMEOUT = 0;
+
+    private CanalConnector   connector;
+
+    /**
+     * 单台client适配器worker的构造方法
+     *
+     * @param canalDestination canal实例名
+     * @param address canal-server地址
+     * @param canalOuterAdapters 外部适配器组
+     */
+    public CanalAdapterWorker(String canalDestination, SocketAddress address,
+                              List<List<CanalOuterAdapter>> canalOuterAdapters){
+        this.canalOuterAdapters = canalOuterAdapters;
+        this.canalDestination = canalDestination;
+        groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
+        connector = CanalConnectors.newSingleConnector(address, canalDestination, "", "");
+    }
+
+    /**
+     * HA模式下client适配器worker的构造方法
+     *
+     * @param canalDestination canal实例名
+     * @param zookeeperHosts zookeeper地址
+     * @param canalOuterAdapters 外部适配器组
+     */
+    public CanalAdapterWorker(String canalDestination, String zookeeperHosts,
+                              List<List<CanalOuterAdapter>> canalOuterAdapters){
+        this.canalOuterAdapters = canalOuterAdapters;
+        this.canalDestination = canalDestination;
+        groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
+        connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", "");
+        ((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT);
+
+        // super.initSwitcher(canalDestination);
+    }
+
+    @Override
+    public void start() {
+        if (!running) {
+            thread = new Thread(new Runnable() {
+
+                @Override
+                public void run() {
+                    process();
+                }
+            });
+            thread.setUncaughtExceptionHandler(handler);
+            thread.start();
+            running = true;
+        }
+    }
+
+    @Override
+    public void stop() {
+        try {
+            if (!running) {
+                return;
+            }
+
+            // if (switcher != null && !switcher.state()) {
+            // switcher.set(true);
+            // }
+
+            connector.stopRunning();
+            running = false;
+
+            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<CanalOuterAdapter> outerAdapters : canalOuterAdapters) {
+                for (CanalOuterAdapter adapter : outerAdapters) {
+                    adapter.destroy();
+                }
+            }
+            logger.info("destination {} all adapters destroyed!", canalDestination);
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+        }
+    }
+
+    private void process() {
+        while (!running)
+            ; // waiting until running == true
+        while (running) {
+            try {
+                // if (switcher != null) {
+                // switcher.get();
+                // }
+                logger.info("=============> Start to connect destination: {} <=============", this.canalDestination);
+                connector.connect();
+                logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination);
+                connector.subscribe();
+                logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination);
+                while (running) {
+                    // try {
+                    // if (switcher != null) {
+                    // switcher.get();
+                    // }
+                    // } catch (TimeoutException e) {
+                    // break;
+                    // }
+
+                    // server配置canal.instance.network.soTimeout(默认: 30s)
+                    // 范围内未与server交互，server将关闭本次socket连接
+                    Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据
+                    long batchId = message.getId();
+                    try {
+                        int size = message.getEntries().size();
+
+                        if (batchId == -1 || size == 0) {
+                            try {
+                                Thread.sleep(1000);
+                            } catch (InterruptedException e) {
+                                // ignore
+                            }
+                        } else {
+                            if (logger.isDebugEnabled()) {
+                                logger.debug("destination: {} batchId: {} batchSize: {} ",
+                                    this.canalDestination,
+                                    batchId,
+                                    size);
+                            }
+                            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);
+                            }
+                        }
+                        connector.ack(batchId); // 提交确认
+                    } catch (Exception e) {
+                        connector.rollback(batchId); // 处理失败, 回滚数据
+                        throw e;
+                    }
+                }
+
+            } catch (Exception e) {
+                logger.error("process error!", e);
+            } finally {
+                connector.disconnect();
+                logger.info("=============> Disconnect destination: {} <=============", this.canalDestination);
+            }
+
+            if (running) { // is reconnect
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    // ignore
+                }
+            }
+
+        }
+    }
+}

client-adapter/launcher/src/main/resources/application.yml

@@ -0,0 +1,21 @@
+server:
+  port: 8081
+
+canal.conf:
+  canalServerHost: 127.0.0.1:11111
+  #zookeeperHosts: slave1:2181
+  #bootstrapServers: localhost:9092 #or rocketmq nameservers:host1:9876;host2:9876
+  flatMessage: false
+
+  canalInstances:
+  - instance: example
+    adapterGroups:
+    - outAdapters:
+      - name: logger
+#      - name: hbase
+#        hosts: slave1:2181
+#        properties: {znodeParent: "/hbase-unsecure"}
+
+adapter.conf.path:
+  adapterConfigs:
+  - hbase/mytest_person2.yml