-
+
+
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java
index 929e9d6a..5922d63e 100644
--- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java
+++ b/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表映射配置加载器
- *
- * 配置统一从hbase-mapping/configs.conf文件作为入口, 该文件包含所有表映射配置的名称或者文件名列表。
- * 每个对应的表配置可以yml配置文件或者以database.table为配置名的简化形式
- *
*
* @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 load() {
logger.info("## Start loading mapping config ... ");
- String mappingConfigContent = readConfigContent(BASE_PATH + "configs.conf");
Map result = new LinkedHashMap<>();
- String[] configLines = mappingConfigContent.split("\n");
- for (String c : configLines) {
+ Collection configs = AdapterConfigs.configs.get("hbase");
+ for (String c : configs) {
if (c == null) {
continue;
}
diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf b/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf
deleted file mode 100644
index ff5b292e..00000000
--- a/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf
+++ /dev/null
@@ -1,5 +0,0 @@
-# 详细映射配置
-mytest_person2.yml
-
-# 简易配置, 只用指定数据库名.表名, 详细配置全部使用默认
-mytest.person
\ No newline at end of file
diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml
similarity index 100%
rename from client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml
rename to client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml
diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml
new file mode 100644
index 00000000..7b94bfc9
--- /dev/null
+++ b/client-adapter/launcher/pom.xml
@@ -0,0 +1,137 @@
+
+
+
+ canal.client-adapter
+ com.alibaba.otter
+ 1.1.1-SNAPSHOT
+
+ 4.0.0
+ com.alibaba.otter
+ launcher
+ jar
+ canal client adapter launcher module for otter ${project.version}
+
+
+
+
+ org.springframework.boot
+ spring-boot-dependencies
+ 2.0.1.RELEASE
+ pom
+ import
+
+
+
+
+
+
+ com.alibaba.otter
+ client-adapter.common
+ ${project.version}
+
+
+ com.alibaba.otter
+ canal.client
+ ${canal_version}
+
+
+ org.yaml
+ snakeyaml
+ 1.19
+
+
+ org.springframework.boot
+ spring-boot-starter-web
+
+
+ org.springframework.boot
+ spring-boot-configuration-processor
+ true
+
+
+
+
+ com.alibaba.otter
+ client-adapter.logger
+ ${project.version}
+
+
+ *
+ *
+
+
+ jar-with-dependencies
+ true
+
+
+ com.alibaba.otter
+ client-adapter.hbase
+ ${project.version}
+
+
+ *
+ *
+
+
+ jar-with-dependencies
+ true
+
+
+
+
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+ 2.0.1.RELEASE
+
+
+
+ repackage
+
+
+
+
+
+ maven-antrun-plugin
+
+
+ package
+
+ run
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-dependency-plugin
+ 2.10
+
+
+ copy-dependencies-to-canal-client-service
+ package
+
+ copy-dependencies
+
+
+ jar-with-dependencies
+ ${project.basedir}/target/lib
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java
new file mode 100644
index 00000000..4b0035d8
--- /dev/null
+++ b/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.");
+ }
+ }
+}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java
new file mode 100644
index 00000000..5b232c36
--- /dev/null
+++ b/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 {
+}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java
new file mode 100644
index 00000000..474d43af
--- /dev/null
+++ b/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 adapterConfigs;
+
+ public List getAdapterConfigs() {
+ return adapterConfigs;
+ }
+
+ public void setAdapterConfigs(List 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);
+ }
+ }
+ }
+ }
+}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java
new file mode 100644
index 00000000..e0912bab
--- /dev/null
+++ b/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> 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> futures = new ArrayList<>();
+ // 组间适配器并行运行
+ for (List outerAdapters : canalOuterAdapters) {
+ final List adapters = outerAdapters;
+ futures.add(groupInnerExecutorService.submit(new Callable() {
+
+ @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 f : futures) {
+ try {
+ if (!f.get()) {
+ logger.error("Outer adapter write failed");
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ // ignore
+ }
+ }
+ }
+ }
+
+ // protected void writeOut(final FlatMessage flatMessage) {
+ // List> futures = new ArrayList<>();
+ // // 组间适配器并行运行
+ // for (List outerAdapters : canalOuterAdapters) {
+ // final List adapters = outerAdapters;
+ // futures.add(groupInnerExecutorService.submit(new Callable() {
+ //
+ // @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 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 outerAdapters : canalOuterAdapters) {
+ for (CanalOuterAdapter adapter : outerAdapters) {
+ adapter.destroy();
+ }
+ }
+ logger.info("topic all connectors destroyed!");
+ }
+
+ public abstract void start();
+
+ public abstract void stop();
+}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java
new file mode 100644
index 00000000..71c9d130
--- /dev/null
+++ b/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 canalWorkers = new HashMap<>();
+
+ private ExtensionLoader 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> canalOuterAdapterGroups = new ArrayList<>();
+
+ for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) {
+ List 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 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");
+ }
+}
diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java
new file mode 100644
index 00000000..7e090c2d
--- /dev/null
+++ b/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> 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> 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 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
+ }
+ }
+
+ }
+ }
+}
diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml
new file mode 100644
index 00000000..bb06e75b
--- /dev/null
+++ b/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
\ No newline at end of file
diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml
index 792ce283..7ca65de1 100644
--- a/client-adapter/pom.xml
+++ b/client-adapter/pom.xml
@@ -2,20 +2,43 @@
-
- canal
- com.alibaba.otter
- 1.1.1-SNAPSHOT
-
4.0.0
com.alibaba.otter
canal.client-adapter
+ 1.1.1-SNAPSHOT
pom
canal client adapter module for otter ${project.version}
+
+
+ UTF-8
+ true
+ true
+ 1.8
+ 1.8
+ UTF-8
+ 1.1.1-SNAPSHOT
+
+
common
logger
hbase
+ launcher
+ example
+
+
+
+ org.apache.maven.plugins
+ maven-compiler-plugin
+ 3.7.0
+
+ ${java_source_version}
+ ${java_target_version}
+ ${file_encoding}
+
+
+
+
diff --git a/client-launcher/src/main/resources/canal-client.yml b/client-launcher/src/main/resources/canal-client.yml
deleted file mode 100644
index d2cb2175..00000000
--- a/client-launcher/src/main/resources/canal-client.yml
+++ /dev/null
@@ -1,24 +0,0 @@
-#canalServerHost: 127.0.0.1:11111
-#zookeeperHosts: slave1:2181
-bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers:host1:9876;host2:9876
-flatMessage: true
-
-#canalInstances:
-#- instance: example
-# adapterGroups:
-# - outAdapters:
-# - name: logger
-# - name: hbase
-# hosts: slave1:2181
-# properties: {znodeParent: "/hbase-unsecure"}
-
-mqTopics:
-- mqMode: rocketmq
- topic: example
- groups:
- - groupId: example
- outAdapters:
- - name: logger
-# - name: hbase
-# hosts: slave1:2181
-# properties: {znodeParent: "/hbase-unsecure"}
diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties
index 81f78b65..25bab241 100644
--- a/deployer/src/main/resources/example/instance.properties
+++ b/deployer/src/main/resources/example/instance.properties
@@ -25,7 +25,7 @@ canal.instance.tsdb.enable=true
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
-#canal.instance.standby.position =
+#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=
diff --git a/deployer/src/main/resources/logback.xml b/deployer/src/main/resources/logback.xml
index d27a1d78..aaaefac7 100644
--- a/deployer/src/main/resources/logback.xml
+++ b/deployer/src/main/resources/logback.xml
@@ -6,7 +6,7 @@
-
+
destination
@@ -16,7 +16,7 @@
../logs/${destination}/${destination}.log
+ class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
../logs/${destination}/%d{yyyy-MM-dd}/${destination}-%d{yyyy-MM-dd}-%i.log.gz
@@ -33,7 +33,7 @@
-
+
destination
@@ -43,7 +43,7 @@
../logs/${destination}/meta.log
+ class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
../logs/${destination}/%d{yyyy-MM-dd}/meta-%d{yyyy-MM-dd}-%i.log.gz
@@ -60,26 +60,26 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+
-
+