diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml index a581ec4c..d61d8d11 100644 --- a/client-adapter/common/pom.xml +++ b/client-adapter/common/pom.xml @@ -28,6 +28,16 @@ joda-time 2.9.4 + + com.alibaba + druid + 1.1.9 + + + mysql + mysql-connector-java + 5.1.40 + diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java similarity index 70% rename from client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java rename to client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java index 5fe0a377..c4ee60fc 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/OuterAdapter.java @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.client.adapter; +import java.util.Map; + import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.SPI; @@ -11,7 +13,7 @@ import com.alibaba.otter.canal.client.adapter.support.SPI; * @version 1.0.0 */ @SPI("logger") -public interface CanalOuterAdapter { +public interface OuterAdapter { /** * 外部适配器初始化接口 @@ -21,11 +23,11 @@ public interface CanalOuterAdapter { void init(CanalOuterAdapterConfiguration configuration); /** - * 往适配器中写入数据 + * 往适配器中同步数据 * * @param dml 数据包 */ - void writeOut(Dml dml); + void sync(Dml dml); // void writeOut(FlatMessage flatMessage); @@ -33,4 +35,14 @@ public interface CanalOuterAdapter { * 外部适配器销毁接口 */ void destroy(); + + /** + * Etl操作 + * + * @param criteriaSql 条件拼接sql + */ + default void etl(String criteriaSql) { + throw new UnsupportedOperationException(); + } + } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java index 13e51af1..44d29ee9 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/AdapterConfigs.java @@ -1,8 +1,28 @@ package com.alibaba.otter.canal.client.adapter.support; -import com.google.common.collect.ArrayListMultimap; -import com.google.common.collect.Multimap; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; -public interface AdapterConfigs { - Multimap configs = ArrayListMultimap.create(); +public class AdapterConfigs { + + private static Map> configs = new ConcurrentHashMap<>(); + + public static void put(String key, String value) { + Set values = configs.get(key); + if (values == null) { + values = new LinkedHashSet<>(); + } + values.add(value); + configs.put(key, values); + } + + public static Set get(String key) { + return configs.get(key); + } + + public static void clear() { + configs.clear(); + } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/DatasourceConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/DatasourceConfig.java new file mode 100644 index 00000000..cf2ec471 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/DatasourceConfig.java @@ -0,0 +1,75 @@ +package com.alibaba.otter.canal.client.adapter.support; + +import com.alibaba.druid.pool.DruidDataSource; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class DatasourceConfig { + + public final static Map DATA_SOURCES = new ConcurrentHashMap<>(); + + private String driver = "com.mysql.jdbc.Driver"; + private String url; + private String database; + private String type = "mysql"; + private String username; + private String password; + private Integer maxActive = 3; + + public String getDriver() { + return driver; + } + + public void setDriver(String driver) { + this.driver = driver; + } + + public String getUrl() { + return url; + } + + public void setUrl(String url) { + this.url = url; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getUsername() { + return username; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Integer getMaxActive() { + return maxActive; + } + + public void setMaxActive(Integer maxActive) { + this.maxActive = maxActive; + } +} diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java index c1f55cd9..cd3f718d 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java @@ -10,7 +10,7 @@ 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.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; @@ -26,10 +26,10 @@ 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 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() { @@ -42,21 +42,21 @@ public abstract class AbstractCanalAdapterWorker { protected void writeOut(final Message message) { List> futures = new ArrayList<>(); // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; futures.add(groupInnerExecutorService.submit(new Callable() { @Override public Boolean call() { try { // 组内适配器穿行运行，尽量不要配置组内适配器 - for (final CanalOuterAdapter c : adapters) { + for (final OuterAdapter c : adapters) { long begin = System.currentTimeMillis(); MessageUtil.parse4Dml(message, new MessageUtil.Consumer() { @Override public void accept(Dml dml) { - c.writeOut(dml); + c.sync(dml); } }); @@ -90,18 +90,18 @@ public abstract class AbstractCanalAdapterWorker { protected void writeOut(final FlatMessage flatMessage) { List> futures = new ArrayList<>(); // 组间适配器并行运行 - for (List outerAdapters : canalOuterAdapters) { - final List adapters = outerAdapters; + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; futures.add(groupInnerExecutorService.submit(new Callable() { @Override public Boolean call() { try { // 组内适配器穿行运行，尽量不要配置组内适配器 - for (CanalOuterAdapter c : adapters) { + for (OuterAdapter c : adapters) { long begin = System.currentTimeMillis(); Dml dml = MessageUtil.flatMessage2Dml(flatMessage); - c.writeOut(dml); + c.sync(dml); if (logger.isDebugEnabled()) { logger.debug("{} elapsed time: {}", c.getClass().getName(), @@ -154,8 +154,8 @@ public abstract class AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("topic connectors' worker thread dead!"); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java index 4429ca80..b2d3e583 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java @@ -4,12 +4,11 @@ import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import org.apache.kafka.clients.consumer.CommitFailedException; import org.apache.kafka.common.errors.WakeupException; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors; import com.alibaba.otter.canal.protocol.FlatMessage; @@ -30,7 +29,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { private boolean flatMessage; public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, - List> canalOuterAdapters, boolean flatMessage){ + List> canalOuterAdapters, boolean flatMessage){ this.canalOuterAdapters = canalOuterAdapters; this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.topic = topic; @@ -81,8 +80,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("topic {} connectors' worker thread dead!", this.topic); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java index 471b6a3d..05ae6d2c 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java @@ -13,7 +13,7 @@ import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; @@ -33,7 +33,7 @@ public class CanalAdapterLoader { private Map canalMQWorker = new HashMap<>(); - private ExtensionLoader loader; + private ExtensionLoader loader; public CanalAdapterLoader(CanalClientConfig canalClientConfig){ this.canalClientConfig = canalClientConfig; @@ -48,7 +48,7 @@ public class CanalAdapterLoader { throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); } - loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, "" /* + loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class, "" /* * TODO * canalClientConfig * . @@ -68,10 +68,10 @@ public class CanalAdapterLoader { // 初始化canal-client的适配器 if (canalClientConfig.getCanalInstances() != null) { for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { - List> canalOuterAdapterGroups = new ArrayList<>(); + List> canalOuterAdapterGroups = new ArrayList<>(); for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { - List canalOutConnectors = new ArrayList<>(); + List canalOutConnectors = new ArrayList<>(); for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) { loadConnector(c, canalOutConnectors); } @@ -93,9 +93,9 @@ public class CanalAdapterLoader { if (canalClientConfig.getMqTopics() != null) { for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) { for (CanalClientConfig.Group group : topic.getGroups()) { - List> canalOuterAdapterGroups = new ArrayList<>(); + List> canalOuterAdapterGroups = new ArrayList<>(); - List canalOuterAdapters = new ArrayList<>(); + List canalOuterAdapters = new ArrayList<>(); for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) { loadConnector(config, canalOuterAdapters); @@ -125,9 +125,9 @@ public class CanalAdapterLoader { } } - private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { + private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { try { - CanalOuterAdapter adapter = loader.getExtension(config.getName()); + OuterAdapter adapter = loader.getExtension(config.getName()); ClassLoader cl = Thread.currentThread().getContextClassLoader(); // 替换ClassLoader Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java index f0f674db..9c126297 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java @@ -8,7 +8,7 @@ import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.CommitFailedException; import org.apache.kafka.common.errors.WakeupException; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; import com.alibaba.otter.canal.protocol.Message; @@ -26,7 +26,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { private String topic; public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, - List> canalOuterAdapters){ + List> canalOuterAdapters){ logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); this.canalOuterAdapters = canalOuterAdapters; this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); diff --git a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java index e341c6a3..e72ed2d5 100644 --- a/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java +++ b/client-adapter/example/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java @@ -6,7 +6,7 @@ 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.adapter.OuterAdapter; import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; import com.alibaba.otter.canal.protocol.Message; @@ -31,7 +31,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, SocketAddress address, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); @@ -46,7 +46,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, String zookeeperHosts, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); @@ -96,8 +96,8 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("destination {} adapters' worker thread dead!", canalDestination); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index d75cea67..c9b67998 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -10,7 +10,7 @@ import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfigLoader; import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService; @@ -25,7 +25,7 @@ import com.alibaba.otter.canal.client.adapter.support.SPI; * @version 1.0.0 */ @SPI("hbase") -public class HbaseAdapter implements CanalOuterAdapter { +public class HbaseAdapter implements OuterAdapter { private static volatile Map mappingConfigCache = null; @@ -83,7 +83,7 @@ public class HbaseAdapter implements CanalOuterAdapter { } @Override - public void writeOut(Dml dml) { + public void sync(Dml dml) { if (dml == null) { return; } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java index 6d70bdde..c3c30c1b 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java @@ -13,9 +13,18 @@ import java.util.Set; * @version 1.0.0 */ public class MappingConfig { + private String dataSourceKey; private HbaseOrm hbaseOrm; + public String getDataSourceKey() { + return dataSourceKey; + } + + public void setDataSourceKey(String dataSourceKey) { + this.dataSourceKey = dataSourceKey; + } + public HbaseOrm getHbaseOrm() { return hbaseOrm; } 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 5922d63e..173fc023 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 @@ -37,7 +37,7 @@ public class MappingConfigLoader { Map result = new LinkedHashMap<>(); - Collection configs = AdapterConfigs.configs.get("hbase"); + Collection configs = AdapterConfigs.get("hbase"); for (String c : configs) { if (c == null) { continue; diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java new file mode 100644 index 00000000..cccd2db6 --- /dev/null +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java @@ -0,0 +1,4 @@ +package com.alibaba.otter.canal.client.adapter.hbase.service; + +public class HbaseEtlService { +} diff --git a/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter similarity index 100% rename from client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter rename to client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter diff --git a/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml index 575a1bf5..f2b6a594 100644 --- a/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml +++ b/client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml @@ -1,8 +1,9 @@ +dataSourceKey: defaultDS hbaseOrm: mode: PHOENIX #NATIVE #STRING database: mytest # 数据库名 table: person2 # 数据库表名 - hbaseTable: MYTEST.PERSON2 # HBase表名 + hbaseTable: MYTEST_PERSON2 # HBase表名 family: CF # 默认统一Family名称 uppercaseQualifier: true # 字段名转大写, 默认为true commitBatch: 3000 # 批量提交的大小 diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 7b94bfc9..8b91f27f 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -81,6 +81,7 @@ + canal-client-launcher org.springframework.boot 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 index 4b0035d8..56bbde8e 100644 --- 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 @@ -1,56 +1,12 @@ 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 index 5b232c36..447f896d 100644 --- 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 @@ -4,7 +4,7 @@ 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 +@ConfigurationProperties(prefix = "canal.conf") public class AdapterCanalConfig extends CanalClientConfig { } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java new file mode 100644 index 00000000..06cb1730 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfig.java @@ -0,0 +1,76 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import java.sql.SQLException; +import java.util.List; +import java.util.Map; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.client.adapter.support.AdapterConfigs; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; + +@Component +@ConfigurationProperties(prefix = "adapter.conf") +public class AdapterConfig { + + private Map datasourceConfigs; + + private List adapterConfigs; + + public List getAdapterConfigs() { + return adapterConfigs; + } + + public Map getDatasourceConfigs() { + return datasourceConfigs; + } + + public void setDatasourceConfigs(Map datasourceConfigs) { + this.datasourceConfigs = datasourceConfigs; + + if (datasourceConfigs != null) { + for (Map.Entry entry : datasourceConfigs.entrySet()) { + DatasourceConfig datasourceConfig = entry.getValue(); + // 加载数据源连接池 + DruidDataSource ds = new DruidDataSource(); + ds.setDriverClassName(datasourceConfig.getDriver()); + ds.setUrl(datasourceConfig.getUrl()); + ds.setUsername(datasourceConfig.getUsername()); + ds.setPassword(datasourceConfig.getPassword()); + ds.setInitialSize(1); + ds.setMinIdle(1); + ds.setMaxActive(datasourceConfig.getMaxActive()); + ds.setMaxWait(60000); + ds.setTimeBetweenEvictionRunsMillis(60000); + ds.setMinEvictableIdleTimeMillis(300000); + ds.setPoolPreparedStatements(false); + ds.setMaxPoolPreparedStatementPerConnectionSize(20); + ds.setValidationQuery("select 1"); + try { + ds.init(); + } catch (SQLException e) { + throw new RuntimeException(e); + } + DatasourceConfig.DATA_SOURCES.put(entry.getKey(), ds); + } + } + } + + public void setAdapterConfigs(List adapterConfigs) { + this.adapterConfigs = adapterConfigs; + + if (adapterConfigs != null) { + AdapterConfigs.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.put(type, ymlFile); + } + } + } + } +} 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 deleted file mode 100644 index 474d43af..00000000 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterConfigPath.java +++ /dev/null @@ -1,35 +0,0 @@ -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 index e0912bab..694251ac 100644 --- 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 @@ -10,7 +10,7 @@ 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.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.MessageUtil; import com.alibaba.otter.canal.protocol.Message; @@ -25,43 +25,33 @@ public abstract class AbstractCanalAdapterWorker { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); protected String canalDestination; // canal实例 - protected List> canalOuterAdapters; // 外部适配器 + 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 Thread.UncaughtExceptionHandler handler = (t, 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() { + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; + futures.add(groupInnerExecutorService.submit(() -> { + try { + // 组内适配器穿行运行，尽量不要配置组内适配器 + for (final OuterAdapter c : adapters) { + long begin = System.currentTimeMillis(); + MessageUtil.parse4Dml(message, c::sync); - @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)); - } + if (logger.isDebugEnabled()) { + logger.debug("{} elapsed time: {}", + c.getClass().getName(), + (System.currentTimeMillis() - begin)); } - return true; - } catch (Exception e) { - return false; } + return true; + } catch (Exception e) { + return false; } })); @@ -131,9 +121,6 @@ public abstract class AbstractCanalAdapterWorker { 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() { @@ -146,8 +133,8 @@ public abstract class AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("topic connectors' worker thread dead!"); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } 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 index 71c9d130..6e7d2bb8 100644 --- 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 @@ -9,7 +9,7 @@ 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.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; import org.slf4j.Logger; @@ -24,15 +24,15 @@ import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; */ public class CanalAdapterLoader { - private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); - private CanalClientConfig canalClientConfig; + private CanalClientConfig canalClientConfig; - private Map canalWorkers = new HashMap<>(); + private Map canalWorkers = new HashMap<>(); - private ExtensionLoader loader; + private ExtensionLoader loader; - public CanalAdapterLoader(CanalClientConfig canalClientConfig) { + public CanalAdapterLoader(CanalClientConfig canalClientConfig){ this.canalClientConfig = canalClientConfig; } @@ -40,12 +40,7 @@ public class CanalAdapterLoader { * 初始化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); + loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class); String canalServerHost = this.canalClientConfig.getCanalServerHost(); SocketAddress sa = null; @@ -58,10 +53,10 @@ public class CanalAdapterLoader { // 初始化canal-client的适配器 if (canalClientConfig.getCanalInstances() != null) { for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { - List> canalOuterAdapterGroups = new ArrayList<>(); + List> canalOuterAdapterGroups = new ArrayList<>(); for (CanalClientConfig.AdapterGroup connectorGroup : instance.getAdapterGroups()) { - List canalOutConnectors = new ArrayList<>(); + List canalOutConnectors = new ArrayList<>(); for (CanalOuterAdapterConfiguration c : connectorGroup.getOutAdapters()) { loadConnector(c, canalOutConnectors); } @@ -80,9 +75,9 @@ public class CanalAdapterLoader { } } - private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { + private void loadConnector(CanalOuterAdapterConfiguration config, List canalOutConnectors) { try { - CanalOuterAdapter adapter = loader.getExtension(config.getName()); + OuterAdapter adapter = loader.getExtension(config.getName()); ClassLoader cl = Thread.currentThread().getContextClassLoader(); // 替换ClassLoader Thread.currentThread().setContextClassLoader(adapter.getClass().getClassLoader()); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java new file mode 100644 index 00000000..f7b9e445 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java @@ -0,0 +1,65 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import javax.annotation.Resource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig; +import com.alibaba.otter.canal.adapter.launcher.config.AdapterConfig; +import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; + +@Component +public class CanalAdapterService { + + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterService.class); + + private static CanalAdapterLoader adapterLoader; + + @Resource + private AdapterCanalConfig adapterCanalConfig; + + // 注入配置保证配置优先先注册 + @Resource + private AdapterConfig adapterConfig; + + @PostConstruct + public void init() { + if (adapterLoader == null) { + try { + logger.info("## start the canal client adapters."); + adapterLoader = new CanalAdapterLoader(adapterCanalConfig); + 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(); + } + for (DruidDataSource druidDataSource : DatasourceConfig.DATA_SOURCES.values()) { + try { + druidDataSource.close(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } 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/loader/CanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterWorker.java index 7e090c2d..1c38b353 100644 --- 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 @@ -6,7 +6,7 @@ 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.adapter.OuterAdapter; import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; import com.alibaba.otter.canal.protocol.Message; @@ -31,7 +31,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, SocketAddress address, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); @@ -46,7 +46,7 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { * @param canalOuterAdapters 外部适配器组 */ public CanalAdapterWorker(String canalDestination, String zookeeperHosts, - List> canalOuterAdapters){ + List> canalOuterAdapters){ this.canalOuterAdapters = canalOuterAdapters; this.canalDestination = canalDestination; groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); @@ -96,8 +96,8 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker { } groupInnerExecutorService.shutdown(); logger.info("destination {} adapters' worker thread dead!", canalDestination); - for (List outerAdapters : canalOuterAdapters) { - for (CanalOuterAdapter adapter : outerAdapters) { + for (List outerAdapters : canalOuterAdapters) { + for (OuterAdapter adapter : outerAdapters) { adapter.destroy(); } } diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index bb06e75b..7adef833 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -16,6 +16,11 @@ canal.conf: # hosts: slave1:2181 # properties: {znodeParent: "/hbase-unsecure"} -adapter.conf.path: +adapter.conf: + datasourceConfigs: + defaultDS: + url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true + username: root + password: 121212 adapterConfigs: - hbase/mytest_person2.yml \ No newline at end of file diff --git a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java index 1885d6c2..27d99aae 100644 --- a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java +++ b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java @@ -3,7 +3,7 @@ package com.alibaba.otter.canal.client.adapter.logger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.SPI; @@ -16,7 +16,7 @@ import com.alibaba.otter.canal.client.adapter.support.SPI; */ @SPI("logger") // logger参数对应CanalOuterAdapterConfiguration配置中的name -public class LoggerAdapterExample implements CanalOuterAdapter { +public class LoggerAdapterExample implements OuterAdapter { private Logger logger = LoggerFactory.getLogger(this.getClass()); @@ -26,7 +26,7 @@ public class LoggerAdapterExample implements CanalOuterAdapter { } @Override - public void writeOut(Dml dml) { + public void sync(Dml dml) { logger.info(dml.toString()); } diff --git a/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter similarity index 100% rename from client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter rename to client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.OuterAdapter