新增adapter-launcher springboot工程

client-adapter/adapter-connector/pom.xml

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>canal.client-adapter</artifactId>
+        <groupId>com.alibaba.otter</groupId>
+        <version>1.1.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>com.alibaba.otter</groupId>
+    <artifactId>adapter-connector</artifactId>
+    <packaging>pom</packaging>
+    <name>canal client adapter connector module for otter ${project.version}</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.otter</groupId>
+            <artifactId>client-adapter.logger</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>*</artifactId>
+                    <groupId>*</groupId>
+                </exclusion>
+            </exclusions>
+            <classifier>jar-with-dependencies</classifier>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.otter</groupId>
+            <artifactId>client-adapter.hbase</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>*</artifactId>
+                    <groupId>*</groupId>
+                </exclusion>
+            </exclusions>
+            <classifier>jar-with-dependencies</classifier>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <version>2.10</version>
+                <executions>
+                    <execution>
+                        <id>copy-dependencies-to-canal-client-service</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>copy-dependencies</goal>
+                        </goals>
+                        <configuration>
+                            <outputDirectory>${project.basedir}/../adapter-launcher/target/lib</outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

client-adapter/adapter-launcher/pom.xml

@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>canal.client-adapter</artifactId>
+        <groupId>com.alibaba.otter</groupId>
+        <version>1.1.1-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>com.alibaba.otter</groupId>
+    <artifactId>adapter-launcher</artifactId>
+    <packaging>jar</packaging>
+    <name>canal client adapter launcher module for otter ${project.version}</name>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>2.0.1.RELEASE</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.otter</groupId>
+            <artifactId>client-adapter.common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.otter</groupId>
+            <artifactId>canal.client</artifactId>
+            <version>${canal_version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.yaml</groupId>
+            <artifactId>snakeyaml</artifactId>
+            <version>1.19</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <version>2.0.1.RELEASE</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <artifactId>maven-antrun-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <tasks>
+                                <copy todir="${project.basedir}/target/config" overwrite="true" >
+                                    <fileset dir="${project.basedir}/src/main/resources" erroronmissingdir="true">
+                                        <include name="*.yml"/>
+                                    </fileset>
+                                </copy>
+                            </tasks>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

client-adapter/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/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/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/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/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/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/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

client-adapter/common/pom.xml

@@ -16,7 +16,7 @@
         <dependency>
             <groupId>com.alibaba.otter</groupId>
             <artifactId>canal.protocol</artifactId>
-            <version>${project.version}</version>
+            <version>${canal_version}</version>
         </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>

client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java

@@ -0,0 +1,8 @@
+package com.alibaba.otter.canal.client.adapter.support;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.Multimap;
+
+public interface AdapterConfigs {
+    Multimap<String, String> configs = ArrayListMultimap.create();
+}

client-adapter/hbase/pom.xml

@@ -22,7 +22,8 @@
         <dependency>
             <groupId>org.yaml</groupId>
             <artifactId>snakeyaml</artifactId>
-            <version>1.17</version>
+            <version>1.19</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.hbase</groupId>
@@ -45,15 +46,23 @@
     <build>
         <plugins>
             <plugin>
-                <artifactId>maven-jar-plugin</artifactId>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <version>2.4</version>
                 <configuration>
-                    <archive>
+                    <descriptorRefs>
-                        <addMavenDescriptor>true</addMavenDescriptor>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
-                    </archive>
+                    </descriptorRefs>
-                    <excludes>
-                        <exclude>**/hbase-mapping/**</exclude>
-                    </excludes>
                 </configuration>
+                <executions>
+                    <execution>
+                        <id>make-assembly</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
             </plugin>
             <plugin>
                 <artifactId>maven-antrun-plugin</artifactId>
@@ -65,9 +74,8 @@
                         </goals>
                         <configuration>
                             <tasks>
-                                <copy todir="${project.basedir}/../../client-launcher/target/canal_client/conf/hbase-mapping" overwrite="true" >
+                                <copy todir="${project.basedir}/../adapter-launcher/target/config/hbase" overwrite="true" >
-                                    <fileset dir="${project.basedir}/src/main/resources/hbase-mapping" erroronmissingdir="true">
+                                    <fileset dir="${project.basedir}/src/main/resources/hbase" erroronmissingdir="true">
-                                        <include name="*.conf"/>
                                         <include name="*.yml"/>
                                     </fileset>
                                 </copy>

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java

@@ -4,6 +4,7 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
@@ -12,12 +13,10 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.yaml.snakeyaml.Yaml;
 
+import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs;
+
 /**
  * HBase表映射配置加载器
- * <p>
- * 配置统一从hbase-mapping/configs.conf文件作为入口, 该文件包含所有表映射配置的名称或者文件名列表。
- * 每个对应的表配置可以yml配置文件或者以database.table为配置名的简化形式
- * </p>
  *
  * @author machengyuan 2018-8-21 下午06:45:49
  * @version 1.0.0
@@ -26,7 +25,7 @@ public class MappingConfigLoader {
 
     private static Logger       logger    = LoggerFactory.getLogger(MappingConfigLoader.class);
 
-    private static final String BASE_PATH = "hbase-mapping/";
+    private static final String BASE_PATH = "hbase";
 
     /**
      * 加载HBase表映射配置
@@ -35,12 +34,11 @@ public class MappingConfigLoader {
      */
     public static Map<String, MappingConfig> load() {
         logger.info("## Start loading mapping config ... ");
-        String mappingConfigContent = readConfigContent(BASE_PATH + "configs.conf");
 
         Map<String, MappingConfig> result = new LinkedHashMap<>();
 
-        String[] configLines = mappingConfigContent.split("\n");
+        Collection<String> configs = AdapterConfigs.configs.get("hbase");
-        for (String c : configLines) {
+        for (String c : configs) {
             if (c == null) {
                 continue;
             }

client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf

@@ -1,5 +0,0 @@
-# 详细映射配置
-mytest_person2.yml
-
-# 简易配置, 只用指定数据库名.表名, 详细配置全部使用默认
-mytest.person

client-adapter/pom.xml

@@ -2,20 +2,43 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>canal</artifactId>
-        <groupId>com.alibaba.otter</groupId>
-        <version>1.1.1-SNAPSHOT</version>
-    </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.alibaba.otter</groupId>
     <artifactId>canal.client-adapter</artifactId>
+    <version>1.1.1-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>canal client adapter module for otter ${project.version}</name>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <maven.test.skip>true</maven.test.skip>
+        <downloadSources>true</downloadSources>
+        <java_source_version>1.8</java_source_version>
+        <java_target_version>1.8</java_target_version>
+        <file_encoding>UTF-8</file_encoding>
+        <canal_version>1.1.1-SNAPSHOT</canal_version>
+    </properties>
+
     <modules>
         <module>common</module>
         <module>logger</module>
         <module>hbase</module>
+        <module>adapter-launcher</module>
+        <module>adapter-connector</module>
     </modules>
 
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.7.0</version>
+                <configuration>
+                    <source>${java_source_version}</source>
+                    <target>${java_target_version}</target>
+                    <encoding>${file_encoding}</encoding>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 </project>

client-launcher/pom.xml

@@ -46,13 +46,6 @@
             <groupId>com.alibaba.otter</groupId>
             <artifactId>client-adapter.logger</artifactId>
             <version>${project.version}</version>
-            <classifier>jar-with-dependencies</classifier>
-        </dependency>
-        <dependency>
-            <groupId>com.alibaba.otter</groupId>
-            <artifactId>client-adapter.hbase</artifactId>
-            <version>${project.version}</version>
-            <classifier>jar-with-dependencies</classifier>
         </dependency>
     </dependencies>
 

deployer/src/main/resources/canal.properties

@@ -63,7 +63,7 @@ canal.instance.parser.parallel = true
 canal.instance.parser.parallelBufferSize = 256
 
 # table meta tsdb info
-canal.instance.tsdb.enable=true
+canal.instance.tsdb.enable=false
 canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
 canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
 canal.instance.tsdb.dbUsername=canal

deployer/src/main/resources/example/instance.properties

@@ -18,7 +18,7 @@ canal.instance.rds.secretkey=
 canal.instance.rds.instanceId=
 
 # table meta tsdb info
-canal.instance.tsdb.enable=true
+canal.instance.tsdb.enable=false
 #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
 #canal.instance.tsdb.dbUsername=canal
 #canal.instance.tsdb.dbPassword=canal
@@ -30,13 +30,13 @@ canal.instance.tsdb.enable=true
 #canal.instance.standby.gtid=
 
 # username/password
-canal.instance.dbUsername=canal
+canal.instance.dbUsername=root
-canal.instance.dbPassword=cZozNf1mzW6EQLGO2q9u99619xbZLO0fbua3EX08r4BWNXb8lAt1aHrTEOBttd6UY8Vnuc0easlVXZDdLtt8BQ==
+canal.instance.dbPassword=121212
 canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
 canal.instance.connectionCharset = UTF-8
 canal.instance.defaultDatabaseName =test
 # enable druid Decrypt database password
-canal.instance.enableDruid=true
+canal.instance.enableDruid=false
 
 # table regex
 canal.instance.filter.regex=.*\\..*

deployer/src/main/resources/logback.xml

@@ -78,8 +78,8 @@
 		<appender-ref ref="CANAL-ROOT" />
 	</logger>
     
-	<root level="WARN">
+	<root level="INFO">
-		<!-- <appender-ref ref="STDOUT"/> -->
+		 <appender-ref ref="STDOUT"/>
-		<appender-ref ref="CANAL-ROOT" />
+		<!--<appender-ref ref="CANAL-ROOT" />-->
 	</root>
 </configuration>

pom.xml

@@ -247,7 +247,7 @@
             <dependency>
                 <groupId>com.alibaba.fastsql</groupId>
                 <artifactId>fastsql</artifactId>
-                <version>2.0.0_preview_630</version>
+                <version>2.0.0_preview_540</version>
             </dependency>
             <dependency>
                 <groupId>com.alibaba</groupId>