From 821f3eced54b8a102fbc25f6ff92946f34fe426e Mon Sep 17 00:00:00 2001 From: duhengforever Date: Fri, 21 Sep 2018 18:41:37 +0800 Subject: [PATCH] Add mq adapter Signed-off-by: duhengforever --- .../adapter/support/CanalClientConfig.java | 9 ++++ .../adapter/loader/CanalAdapterLoader.java | 50 ++++++++++--------- .../src/main/resources/canal-client.yml | 2 + deployer/src/main/resources/mq.yml | 2 +- .../canal/rocketmq/CanalRocketMQProducer.java | 5 +- 5 files changed, 40 insertions(+), 28 deletions(-) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index e0658138..6619555c 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -111,11 +111,20 @@ public class CanalClientConfig { } public static class MQTopic { + private String mqMode; private String topic; private List groups = new ArrayList<>(); + public String getMqMode() { + return mqMode; + } + + public void setMqMode(String mqMode) { + this.mqMode = mqMode; + } + public String getTopic() { return topic; } diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java index a705fad7..84517525 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java @@ -4,6 +4,7 @@ import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; 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; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -17,7 +18,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** - * RocktMQ外部适配器的加载器 + * MQ外部适配器的加载器 * * @version 1.0.0 */ @@ -29,8 +30,7 @@ public class CanalAdapterLoader { private Map canalWorkers = new HashMap<>(); - private Map canalKafkaWorkers = new HashMap<>(); - private Map canalMQWorker = new HashMap<>(); + private Map canalMQWorker = new HashMap<>(); private ExtensionLoader loader; @@ -42,7 +42,7 @@ public class CanalAdapterLoader { * 初始化canal-client、 canal-client-rocketmq的适配器 */ public void init() { - // canal instances 和 rocketmq topics 配置不能同时为空 + // canal instances 和 mq topics 配置不能同时为空 if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) { throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); } @@ -58,11 +58,6 @@ public class CanalAdapterLoader { } String zkHosts = this.canalClientConfig.getZookeeperHosts(); - // if (zkHosts == null && sa == null) { - // throw new RuntimeException("Blank config property: canalServerHost or - // zookeeperHosts"); - // } - // 初始化canal-client的适配器 if (canalClientConfig.getCanalInstances() != null) { for (CanalClientConfig.CanalInstance instance : canalClientConfig.getCanalInstances()) { @@ -96,19 +91,27 @@ public class CanalAdapterLoader { List canalOuterAdapters = new ArrayList<>(); for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) { - // for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) { loadConnector(config, canalOuterAdapters); - // } } canalOuterAdapterGroups.add(canalOuterAdapters); - CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker( - canalClientConfig.getBootstrapServers(), - topic.getTopic(), - group.getGroupId(), - canalOuterAdapterGroups); - canalMQWorker.put(topic.getTopic() + "-" + group.getGroupId(), rocketMQWorker); - rocketMQWorker.start(); + if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) { + CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker( + canalClientConfig.getBootstrapServers(), + topic.getTopic(), + group.getGroupId(), + canalOuterAdapterGroups); + canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); + rocketMQWorker.start(); + } else if ("kafka".equalsIgnoreCase(topic.getMqMode())) { + CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkHosts, + canalClientConfig.getBootstrapServers(), + topic.getTopic(), + group.getGroupId(), + canalOuterAdapterGroups); + canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker); + canalKafkaWorker.start(); + } logger.info("Start adapter for canal-client rocketmq topic: {} succeed", topic.getTopic() + "-" + group.getGroupId()); } @@ -149,15 +152,14 @@ public class CanalAdapterLoader { } stopExecutorService.shutdown(); } - if (canalKafkaWorkers.size() > 0) { - ExecutorService stopKafkaExecutorService = Executors.newFixedThreadPool(canalKafkaWorkers.size()); - for (CanalAdapterKafkaWorker v : canalKafkaWorkers.values()) { - final CanalAdapterKafkaWorker cakw = v; + if (canalMQWorker.size() > 0) { + ExecutorService stopKafkaExecutorService = Executors.newFixedThreadPool(canalMQWorker.size()); + for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) { + final AbstractCanalAdapterWorker worker = tmp; stopKafkaExecutorService.submit(new Runnable() { - @Override public void run() { - cakw.stop(); + worker.stop(); } }); } diff --git a/client-launcher/src/main/resources/canal-client.yml b/client-launcher/src/main/resources/canal-client.yml index a5fc4813..6117fb95 100644 --- a/client-launcher/src/main/resources/canal-client.yml +++ b/client-launcher/src/main/resources/canal-client.yml @@ -10,7 +10,9 @@ canalServerHost: 127.0.0.1:11111 # - name: hbase # hosts: slave1:2181 # properties: {znodeParent: "/hbase-unsecure"} + mqTopics: +- mqMode: rocketmq - topic: example groups: - groupId: example_g1 diff --git a/deployer/src/main/resources/mq.yml b/deployer/src/main/resources/mq.yml index d8b18f4a..214bcb6f 100644 --- a/deployer/src/main/resources/mq.yml +++ b/deployer/src/main/resources/mq.yml @@ -10,7 +10,7 @@ filterTransactionEntry: true canalDestinations: - canalDestination: example topic: example - partition: + partition: 0 # 一个destination可以对应多个topic #topics: # - topic: example diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index 5a2900de..b7c57f2a 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -35,14 +35,13 @@ public class CanalRocketMQProducer implements CanalMQProducer { } @Override - public void send(MQProperties.Topic topic, com.alibaba.otter.canal.protocol.Message data) { + public void send(final MQProperties.Topic topic, com.alibaba.otter.canal.protocol.Message data) { try { Message message = new Message(topic.getTopic(), CanalMessageSerializer.serializer(data)); this.defaultMQProducer.send(message, new MessageQueueSelector() { @Override public MessageQueue select(List mqs, Message msg, Object arg) { - // int index = (arg.hashCode() % mqs.size()); - return mqs.get(1); + return mqs.get(topic.getPartition()); } }, null); } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {