From 09df602f3bd060a297bb7362409b344523cb5b8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Thu, 20 Sep 2018 11:30:10 +0800 Subject: [PATCH 001/328] support filterTransactionEntry --- .../parse/inbound/mysql/MysqlEventParser.java | 25 ++++++----- .../otter/canal/protocol/FlatMessage.java | 9 +++- .../otter/canal/kafka/CanalKafkaProducer.java | 2 - .../otter/canal/kafka/MessageSerializer.java | 7 ++-- .../server/netty/handler/SessionHandler.java | 3 +- .../canal/sink/entry/EntryEventSink.java | 42 ++++++++++--------- 6 files changed, 47 insertions(+), 41 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index a5c8d870..cf102d3d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -8,7 +8,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.TimerTask; -import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.lang.StringUtils; import org.springframework.util.CollectionUtils; @@ -49,27 +48,27 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE private CanalHAController haController = null; - private int defaultConnectionTimeoutInSeconds = 30; // sotimeout + private int defaultConnectionTimeoutInSeconds = 30; // sotimeout private int receiveBufferSize = 64 * 1024; private int sendBufferSize = 64 * 1024; // 数据库信息 - protected AuthenticationInfo masterInfo; // 主库 - protected AuthenticationInfo standbyInfo; // 备库 + protected AuthenticationInfo masterInfo; // 主库 + protected AuthenticationInfo standbyInfo; // 备库 // binlog信息 protected EntryPosition masterPosition; protected EntryPosition standbyPosition; - private long slaveId; // 链接到mysql的slave + private long slaveId; // 链接到mysql的slave // 心跳检查信息 - private String detectingSQL; // 心跳sql - private MysqlConnection metaConnection; // 查询meta信息的链接 - private TableMetaCache tableMetaCache; // 对应meta - private int fallbackIntervalInSeconds = 60; // 切换回退时间 - private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验 - private BinlogImage[] supportBinlogImages; // 支持的binlogImage,如果设置会执行强校验 + private String detectingSQL; // 心跳sql + private MysqlConnection metaConnection; // 查询meta信息的链接 + private TableMetaCache tableMetaCache; // 对应meta + private int fallbackIntervalInSeconds = 60; // 切换回退时间 + private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验 + private BinlogImage[] supportBinlogImages; // 支持的binlogImage,如果设置会执行强校验 // update by yishun.chen,特殊异常处理参数 - private int dumpErrorCount = 0; // binlogDump失败异常计数 - private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 + private int dumpErrorCount = 0; // binlogDump失败异常计数 + private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 protected ErosaConnection buildErosaConnection() { return buildMysqlConnection(this.runningInfo); diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index e6aab531..d7150b79 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -1,7 +1,12 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; -import java.util.*; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; import com.google.common.collect.Table; import com.google.protobuf.ByteString; @@ -243,6 +248,8 @@ public class FlatMessage implements Serializable { } int hash = value.hashCode(); int pkHash = Math.abs(hash) % partitionsNum; + // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序 + pkHash = Math.abs(pkHash); FlatMessage flatMessageTmp = partitionMessages[pkHash]; if (flatMessageTmp == null) { diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 84656fb1..4942a37b 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -2,12 +2,10 @@ package com.alibaba.otter.canal.kafka; import java.util.List; import java.util.Properties; -import java.util.concurrent.Future; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java index 49db5aa1..86b31502 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java @@ -28,6 +28,7 @@ public class MessageSerializer implements Serializer { } @Override + @SuppressWarnings("deprecation") public byte[] serialize(String topic, Message data) { try { if (data != null) { @@ -47,11 +48,9 @@ public class MessageSerializer implements Serializer { messageSize += 1 * rowEntries.size(); // packet size int size = 0; - size += CodedOutputStream.computeEnumSize(3, - PacketType.MESSAGES.getNumber()); + size += CodedOutputStream.computeEnumSize(3, PacketType.MESSAGES.getNumber()); size += CodedOutputStream.computeTagSize(5) - + CodedOutputStream.computeRawVarint32Size(messageSize) - + messageSize; + + CodedOutputStream.computeRawVarint32Size(messageSize) + messageSize; // build data byte[] body = new byte[size]; CodedOutputStream output = CodedOutputStream.newInstance(body); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java index fe8c5538..d060c026 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java @@ -3,7 +3,6 @@ package com.alibaba.otter.canal.server.netty.handler; import java.util.List; import java.util.concurrent.TimeUnit; -import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.jboss.netty.buffer.ChannelBuffer; @@ -34,6 +33,7 @@ import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; import com.alibaba.otter.canal.server.netty.NettyUtils; +import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator; import com.google.protobuf.ByteString; import com.google.protobuf.CodedOutputStream; import com.google.protobuf.WireFormat; @@ -50,6 +50,7 @@ public class SessionHandler extends SimpleChannelHandler { this.embeddedServer = embeddedServer; } + @SuppressWarnings("deprecation") public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { logger.info("message receives in session handler..."); long start = System.nanoTime(); diff --git a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java index 7afde6b6..76124ec4 100644 --- a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java +++ b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java @@ -33,10 +33,13 @@ public class EntryEventSink extends AbstractCanalEventSink eventStore; - protected boolean filterTransactionEntry = false; // 是否需要过滤事务头/尾 + protected boolean filterTransactionEntry = false; // 是否需要尽可能过滤事务头/尾 protected boolean filterEmtryTransactionEntry = true; // 是否需要过滤空的事务头/尾 protected long emptyTransactionInterval = 5 * 1000; // 空的事务输出的频率 - protected long emptyTransctionThresold = 8192; // 超过1024个事务头，输出一个 + protected long emptyTransctionThresold = 8192; // 超过8192个事务头，输出一个 + + protected volatile long lastTransactionTimestamp = 0L; + protected AtomicLong lastTransactionCount = new AtomicLong(0L); protected volatile long lastEmptyTransactionTimestamp = 0L; protected AtomicLong lastEmptyTransactionCount = new AtomicLong(0L); private AtomicLong eventsSinkBlockingTime = new AtomicLong(0L); @@ -74,17 +77,7 @@ public class EntryEventSink extends AbstractCanalEventSink entrys, InetSocketAddress remoteAddress, String destination) throws CanalSinkException, InterruptedException { - List rowDatas = entrys; - if (filterTransactionEntry) { - rowDatas = new ArrayList(); - for (CanalEntry.Entry entry : entrys) { - if (entry.getEntryType() == EntryType.ROWDATA) { - rowDatas.add(entry); - } - } - } - - return sinkData(rowDatas, remoteAddress); + return sinkData(entrys, remoteAddress); } private boolean sinkData(List entrys, InetSocketAddress remoteAddress) @@ -97,17 +90,26 @@ public class EntryEventSink extends AbstractCanalEventSink emptyTransactionInterval + || lastTransactionCount.incrementAndGet() > emptyTransctionThresold) { + lastTransactionCount.set(0L); + lastTransactionTimestamp = currentTimestamp; + continue; + } + } + hasRowData |= (entry.getEntryType() == EntryType.ROWDATA); hasHeartBeat |= (entry.getEntryType() == EntryType.HEARTBEAT); + Event event = new Event(new LogIdentity(remoteAddress, -1L), entry); + events.add(event); } - if (hasRowData) { - // 存在row记录 - return doSink(events); - } else if (hasHeartBeat) { - // 存在heartbeat记录，直接跳给后续处理 + if (hasRowData || hasHeartBeat) { + // 存在row记录 或者 存在heartbeat记录，直接跳给后续处理 return doSink(events); } else { // 需要过滤的数据 From d5aa324ce455471cf2ea62d519cecf0062628001 Mon Sep 17 00:00:00 2001 From: duhengforever Date: Thu, 20 Sep 2018 15:48:52 +0800 Subject: [PATCH 002/328] Add RocketMQ canal connector --- .../adapter/support/CanalClientConfig.java | 12 +- .../adapter/loader/CanalAdapterLoader.java | 43 ++-- .../loader/CanalAdapterRocketMQWorker.java | 165 +++++++++++++ .../src/main/resources/canal-client.yml | 26 +-- client/pom.xml | 217 +++++++++--------- .../otter/canal/client/DeserializerUtil.java | 42 ++++ .../client/kafka/MessageDeserializer.java | 39 +--- .../client/rocketmq/ConsumerBatchMessage.java | 47 ++++ .../rocketmq/RocketMQCanalConnector.java | 197 ++++++++++++++++ .../RocketMQCanalConnectorProvider.java | 20 ++ .../rocketmq/RocketMQCanalListener.java | 11 + deployer/pom.xml | 2 +- .../otter/canal/deployer/CanalLauncher.java | 25 +- deployer/src/main/resources/canal.properties | 4 +- .../resources/example/instance.properties | 9 +- .../src/main/resources/{kafka.yml => mq.yml} | 2 +- server/pom.xml | 5 + .../canal/common/CanalMessageSerializer.java | 70 ++++++ .../MQProperties.java} | 12 +- .../otter/canal/kafka/CanalKafkaProducer.java | 39 +--- .../otter/canal/kafka/MessageSerializer.java | 56 +---- .../canal/rocketmq/CanalRocketMQProducer.java | 58 +++++ .../CanalMQStarter.java} | 80 +++---- .../otter/canal/spi/CanalMQProducer.java | 28 +++ 24 files changed, 881 insertions(+), 328 deletions(-) create mode 100644 client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/DeserializerUtil.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnectorProvider.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalListener.java rename deployer/src/main/resources/{kafka.yml => mq.yml} (85%) create mode 100644 server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java rename server/src/main/java/com/alibaba/otter/canal/{kafka/KafkaProperties.java => common/MQProperties.java} (92%) create mode 100644 server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java rename server/src/main/java/com/alibaba/otter/canal/{kafka/CanalKafkaStarter.java => server/CanalMQStarter.java} (66%) create mode 100644 server/src/main/java/com/alibaba/otter/canal/spi/CanalMQProducer.java 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 f6fa7707..e0658138 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 @@ -20,7 +20,7 @@ public class CanalClientConfig { private String bootstrapServers; - private List kafkaTopics; + private List mqTopics; private List canalInstances; @@ -56,12 +56,12 @@ public class CanalClientConfig { this.bootstrapServers = bootstrapServers; } - public List getKafkaTopics() { - return kafkaTopics; + public List getMqTopics() { + return mqTopics; } - public void setKafkaTopics(List kafkaTopics) { - this.kafkaTopics = kafkaTopics; + public void setMqTopics(List mqTopics) { + this.mqTopics = mqTopics; } public List getCanalInstances() { @@ -110,7 +110,7 @@ public class CanalClientConfig { } } - public static class KafkaTopic { + public static class MQTopic { private String 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 98973f11..a705fad7 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 @@ -17,34 +17,34 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** - * 外部适配器的加载器 + * RocktMQ外部适配器的加载器 * - * @author machengyuan 2018-8-19 下午11:45:49 * @version 1.0.0 */ 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 Map canalKafkaWorkers = new HashMap<>(); + private Map canalMQWorker = new HashMap<>(); - private ExtensionLoader loader; + private ExtensionLoader loader; - public CanalAdapterLoader(CanalClientConfig canalClientConfig){ + public CanalAdapterLoader(CanalClientConfig canalClientConfig) { this.canalClientConfig = canalClientConfig; } /** - * 初始化canal-client、 canal-client-kafka的适配器 + * 初始化canal-client、 canal-client-rocketmq的适配器 */ public void init() { - // canal instances 和 kafka topics 配置不能同时为空 - if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getKafkaTopics() == null) { - throw new RuntimeException("Blank config property: canalInstances or canalKafkaTopics"); + // canal instances 和 rocketmq topics 配置不能同时为空 + if (canalClientConfig.getCanalInstances() == null && canalClientConfig.getMqTopics() == null) { + throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); } loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, @@ -87,10 +87,10 @@ public class CanalAdapterLoader { } } - // 初始化canal-client-kafka的适配器 - if (canalClientConfig.getKafkaTopics() != null) { - for (CanalClientConfig.KafkaTopic kafkaTopic : canalClientConfig.getKafkaTopics()) { - for (CanalClientConfig.Group group : kafkaTopic.getGroups()) { + // 初始化canal-client-rocketmq的适配器 + if (canalClientConfig.getMqTopics() != null) { + for (CanalClientConfig.MQTopic topic : canalClientConfig.getMqTopics()) { + for (CanalClientConfig.Group group : topic.getGroups()) { List> canalOuterAdapterGroups = new ArrayList<>(); List canalOuterAdapters = new ArrayList<>(); @@ -102,16 +102,15 @@ public class CanalAdapterLoader { } canalOuterAdapterGroups.add(canalOuterAdapters); - // String zkServers = canalClientConfig.getZookeeperHosts(); - CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkHosts, + CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker( canalClientConfig.getBootstrapServers(), - kafkaTopic.getTopic(), + topic.getTopic(), group.getGroupId(), canalOuterAdapterGroups); - canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker); - canalKafkaWorker.start(); - logger.info("Start adapter for canal-client kafka topic: {} succeed", - kafkaTopic.getTopic() + "-" + group.getGroupId()); + canalMQWorker.put(topic.getTopic() + "-" + group.getGroupId(), rocketMQWorker); + rocketMQWorker.start(); + logger.info("Start adapter for canal-client rocketmq topic: {} succeed", + topic.getTopic() + "-" + group.getGroupId()); } } } diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java new file mode 100644 index 00000000..029d9ada --- /dev/null +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java @@ -0,0 +1,165 @@ +package com.alibaba.otter.canal.client.adapter.loader; + +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; +import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; +import com.alibaba.otter.canal.protocol.Message; +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; + +/** + * kafka对应的client适配器工作线程 + * + * @author machengyuan 2018-8-19 下午11:30:49 + * @version 1.0.0 + */ +public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { + + private RocketMQCanalConnector connector; + + private String topic; + + public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, + List> canalOuterAdapters) { + this.canalOuterAdapters = canalOuterAdapters; + this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); + this.topic = topic; + this.canalDestination = topic; + connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId); + } + + @Override + public void start() { + if (!running) { + thread = new Thread(new Runnable() { + + @Override + public void run() { + process(); + } + }); + thread.setUncaughtExceptionHandler(handler); + running = true; + thread.start(); + } + } + + @Override + public void stop() { + try { + if (!running) { + return; + } + + connector.stopRunning(); + running = false; + + // if (switcher != null && !switcher.state()) { + // switcher.set(true); + // } + + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + groupInnerExecutorService.shutdown(); + logger.info("topic {} connectors' worker thread dead!", this.topic); + for (List outerAdapters : canalOuterAdapters) { + for (CanalOuterAdapter adapter : outerAdapters) { + adapter.destroy(); + } + } + logger.info("topic {} all connectors destroyed!", this.topic); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private void process() { + while (!running) + ; + ExecutorService executor = Executors.newSingleThreadExecutor(); + final AtomicBoolean executing = new AtomicBoolean(true); + while (running) { + try { + logger.info("=============> Start to connect topic: {} <=============", this.topic); + connector.connect(); + logger.info("=============> Start to subscribe topic: {}<=============", this.topic); + connector.subscribe(); + logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); + while (running) { + try { + // switcher.get(); //等待开关开启 + + final Message message = connector.getWithoutAck(1); + + executing.set(true); + if (message != null) { + executor.submit(new Runnable() { + + @Override + public void run() { + try { + 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); + } + connector.ack(message.getId()); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } finally { + executing.compareAndSet(true, false); + } + } + }); + } else { + logger.debug("Message is null"); + } + } catch (CommitFailedException e) { + logger.warn(e.getMessage()); + } catch (Exception e) { + logger.error(e.getMessage(), e); + TimeUnit.SECONDS.sleep(1L); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + executor.shutdown(); + + try { + connector.unsubscribe(); + } catch (WakeupException e) { + // No-op. Continue process + } + connector.stopRunning(); + logger.info("=============> Disconnect topic: {} <=============", this.topic); + } +} diff --git a/client-launcher/src/main/resources/canal-client.yml b/client-launcher/src/main/resources/canal-client.yml index 12e016a7..a5fc4813 100644 --- a/client-launcher/src/main/resources/canal-client.yml +++ b/client-launcher/src/main/resources/canal-client.yml @@ -2,20 +2,20 @@ canalServerHost: 127.0.0.1:11111 #zookeeperHosts: slave1:2181 #bootstrapServers: slave1:6667,slave2:6667 -canalInstances: -- instance: example - adapterGroups: - - outAdapters: - - name: logger - - name: hbase - hosts: slave1:2181 - properties: {znodeParent: "/hbase-unsecure"} -#kafkaTopics: -#- topic: example -# groups: -# - groupId: example_g1 -# outAdapters: +#canalInstances: +#- instance: example +# adapterGroups: +# - outAdapters: # - name: logger # - name: hbase # hosts: slave1:2181 # properties: {znodeParent: "/hbase-unsecure"} +mqTopics: +- topic: example + groups: + - groupId: example_g1 + outAdapters: + - name: logger +# - name: hbase +# hosts: slave1:2181 +# properties: {znodeParent: "/hbase-unsecure"} diff --git a/client/pom.xml b/client/pom.xml index 2e48f9c5..b44e3129 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -1,108 +1,115 @@ - - 4.0.0 - - com.alibaba.otter - canal - 1.1.1-SNAPSHOT - ../pom.xml - - com.alibaba.otter - canal.client - jar - canal client module for otter ${project.version} - - - com.alibaba.otter - canal.protocol - ${project.version} - - - - org.apache.kafka - kafka-clients - 1.1.1 - provided - + + 4.0.0 + + com.alibaba.otter + canal + 1.1.1-SNAPSHOT + ../pom.xml + + com.alibaba.otter + canal.client + jar + canal client module for otter ${project.version} + + + com.alibaba.otter + canal.protocol + ${project.version} + + + + org.apache.kafka + kafka-clients + 1.1.1 + provided + + + org.apache.rocketmq + rocketmq-client + 4.3.0 + + + + junit + junit + + - - - junit - junit - - - - - - dev - - true - - env - !javadoc - - - + + + dev + + true + + env + !javadoc + + + - - javadoc - - - env - javadoc - - - - - - org.apache.maven.plugins - maven-javadoc-plugin - 2.9.1 - - - attach-javadocs - package - - jar - - - - - true - public - true -
${project.artifactId}-${project.version}
-
${project.artifactId}-${project.version}
- ${project.artifactId}-${project.version} - - https://github.com/alibaba/canal - - ${project.build.directory}/apidocs/apidocs/${project.version} - - - - org.apache.maven.plugins - maven-scm-publish-plugin - 1.0-beta-2 - - - attach-javadocs - package - - publish-scm - - - - - ${project.build.directory}/scmpublish - Publishing javadoc for ${project.artifactId}:${project.version} - ${project.build.directory}/apidocs - true - scm:git:git@github.com:alibaba/canal.git - gh-pages - - - - - - + + javadoc + + + env + javadoc + + + + + + org.apache.maven.plugins + maven-javadoc-plugin + 2.9.1 + + + attach-javadocs + package + + jar + + + + + true + public + true +
${project.artifactId}-${project.version}
+
${project.artifactId}-${project.version}
+ ${project.artifactId}-${project.version} + + https://github.com/alibaba/canal + + ${project.build.directory}/apidocs/apidocs/${project.version} + + + + + org.apache.maven.plugins + maven-scm-publish-plugin + 1.0-beta-2 + + + attach-javadocs + package + + publish-scm + + + + + ${project.build.directory}/scmpublish + Publishing javadoc for ${project.artifactId}:${project.version} + + ${project.build.directory}/apidocs + true + scm:git:git@github.com:alibaba/canal.git + gh-pages + + + + + + diff --git a/client/src/main/java/com/alibaba/otter/canal/client/DeserializerUtil.java b/client/src/main/java/com/alibaba/otter/canal/client/DeserializerUtil.java new file mode 100644 index 00000000..54b1bb7a --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/DeserializerUtil.java @@ -0,0 +1,42 @@ +package com.alibaba.otter.canal.client; + +import com.alibaba.otter.canal.protocol.CanalEntry; +import com.alibaba.otter.canal.protocol.CanalPacket; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.protocol.exception.CanalClientException; +import com.google.protobuf.ByteString; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DeserializerUtil { + private static Logger logger = LoggerFactory.getLogger(DeserializerUtil.class); + + public static Message deserializer(byte[] data) { + try { + if (data == null) { + return null; + } else { + CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data); + switch (p.getType()) { + case MESSAGES: { + if (!p.getCompression().equals(CanalPacket.Compression.NONE)) { + throw new CanalClientException("compression is not supported in this connector"); + } + + CanalPacket.Messages messages = CanalPacket.Messages.parseFrom(p.getBody()); + Message result = new Message(messages.getBatchId()); + for (ByteString byteString : messages.getMessagesList()) { + result.addEntry(CanalEntry.Entry.parseFrom(byteString)); + } + return result; + } + default: + break; + } + } + } catch (Exception e) { + logger.error("Error when deserializing byte[] to message ", e); + } + return null; + } +} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java index e716d2d6..af381283 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java @@ -1,17 +1,12 @@ package com.alibaba.otter.canal.client.kafka; +import com.alibaba.otter.canal.client.DeserializerUtil; +import com.alibaba.otter.canal.protocol.Message; import java.util.Map; - import org.apache.kafka.common.serialization.Deserializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.otter.canal.protocol.CanalEntry; -import com.alibaba.otter.canal.protocol.CanalPacket; -import com.alibaba.otter.canal.protocol.Message; -import com.alibaba.otter.canal.protocol.exception.CanalClientException; -import com.google.protobuf.ByteString; - /** * Kafka Message类的反序列化 * @@ -27,34 +22,8 @@ public class MessageDeserializer implements Deserializer { } @Override - public Message deserialize(String topic, byte[] data) { - try { - if (data == null) { - return null; - } - else { - CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data); - switch (p.getType()) { - case MESSAGES: { - if (!p.getCompression().equals(CanalPacket.Compression.NONE)) { - throw new CanalClientException("compression is not supported in this connector"); - } - - CanalPacket.Messages messages = CanalPacket.Messages.parseFrom(p.getBody()); - Message result = new Message(messages.getBatchId()); - for (ByteString byteString : messages.getMessagesList()) { - result.addEntry(CanalEntry.Entry.parseFrom(byteString)); - } - return result; - } - default: - break; - } - } - } catch (Exception e) { - logger.error("Error when deserializing byte[] to message "); - } - return null; + public Message deserialize(String topic1, byte[] data) { + return DeserializerUtil.deserializer(data); } @Override diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java new file mode 100644 index 00000000..ef83a798 --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java @@ -0,0 +1,47 @@ +package com.alibaba.otter.canal.client.rocketmq; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +public class ConsumerBatchMessage { + private final BlockingQueue data; + private CountDownLatch latch; + private boolean hasFailure = false; + + public ConsumerBatchMessage(BlockingQueue data) { + this.data = data; + latch = new CountDownLatch(data.size()); + } + + public boolean waitFinish(long timeout) throws InterruptedException { + return latch.await(timeout, TimeUnit.MILLISECONDS); + } + + public boolean isSuccess() { + return !hasFailure; + } + + public BlockingQueue getData() { + return data; + } + + /** + * Countdown if the sub message is successful. + */ + public void ack() { + latch.countDown(); + } + + /** + * Countdown and fail-fast if the sub message is failed. + */ + public void fail() { + hasFailure = true; + // fail fast + long count = latch.getCount(); + for (int i = 0; i < count; i++) { + latch.countDown(); + } + } +} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java new file mode 100644 index 00000000..e6407308 --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java @@ -0,0 +1,197 @@ +package com.alibaba.otter.canal.client.rocketmq; + +import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.client.DeserializerUtil; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.protocol.exception.CanalClientException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.apache.commons.lang.StringUtils; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus; +import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.message.MessageExt; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class RocketMQCanalConnector implements CanalConnector { + private static final Logger logger = LoggerFactory.getLogger(RocketMQCanalConnector.class); + + private String nameServer; + private String topic; + private String groupName; + private volatile boolean connected = false; + private DefaultMQPushConsumer rocketMQConsumer; + private BlockingQueue> messageBlockingQueue; + Map> messageCache; + private long batchProcessTimeout = 10000; + + public RocketMQCanalConnector(String nameServer, String topic, String groupName) { + this.nameServer = nameServer; + this.topic = topic; + this.groupName = groupName; + messageBlockingQueue = new LinkedBlockingQueue<>(); + messageCache = new ConcurrentHashMap<>(); + } + + @Override + public void connect() throws CanalClientException { + rocketMQConsumer = new DefaultMQPushConsumer(groupName); + } + + @Override + public void disconnect() throws CanalClientException { + rocketMQConsumer.shutdown(); + } + + @Override + public boolean checkValid() throws CanalClientException { + return connected; + } + + @Override + public synchronized void subscribe(String filter) throws CanalClientException { + if (connected) { + return; + } + try { + rocketMQConsumer.subscribe(topic, "*"); + if (rocketMQConsumer == null) { + rocketMQConsumer = new DefaultMQPushConsumer(groupName); + if (!StringUtils.isBlank(nameServer)) { + rocketMQConsumer.setNamesrvAddr(nameServer); + } + } + rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() { + @Override + public ConsumeOrderlyStatus consumeMessage(List messageExts, + ConsumeOrderlyContext context) { + context.setAutoCommit(true); + boolean isSuccess = process(messageExts); + if (isSuccess) { + return ConsumeOrderlyStatus.SUCCESS; + } else { + return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT; + } + } + }); + rocketMQConsumer.start(); + } catch (MQClientException ex) { + connected = false; + logger.error("Start RocketMQ consumer error", ex); + } + connected = true; + } + + private boolean process(List messageExts) { + BlockingQueue messageList = new LinkedBlockingQueue<>(); + for (MessageExt messageExt : messageExts) { + byte[] data = messageExt.getBody(); + Message message = DeserializerUtil.deserializer(data); + try { + messageList.put(message); + } catch (InterruptedException ex) { + logger.error("Add message error"); + } + } + ConsumerBatchMessage batchMessage = new ConsumerBatchMessage<>(messageList); + try { + messageBlockingQueue.put(batchMessage); + } catch (InterruptedException e) { + logger.error("Put message to queue error", e); + throw new RuntimeException(e); + } + boolean isCompleted; + try { + isCompleted = batchMessage.waitFinish(batchProcessTimeout); + } catch (InterruptedException e) { + logger.error("Interrupted when waiting messages to be finished.", e); + throw new RuntimeException(e); + } + boolean isSuccess = batchMessage.isSuccess(); + return isCompleted && isSuccess; + } + + @Override + public void subscribe() throws CanalClientException { + this.subscribe(null); + } + + @Override + public void unsubscribe() throws CanalClientException { + this.rocketMQConsumer.unsubscribe(this.topic); + } + + @Override + public Message get(int batchSize) throws CanalClientException { + Message message = getWithoutAck(batchSize); + ack(message.getId()); + return message; + } + + @Override + public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + Message message = getWithoutAck(batchSize, timeout, unit); + ack(message.getId()); + return message; + } + + private Message getMessage(ConsumerBatchMessage consumerBatchMessage) { + BlockingQueue messageList = consumerBatchMessage.getData(); + if (messageList != null & messageList.size() > 0) { + Message message = messageList.poll(); + messageCache.put(message.getId(), consumerBatchMessage); + return message; + } + return null; + } + + @Override + public Message getWithoutAck(int batchSize) throws CanalClientException { + ConsumerBatchMessage batchMessage = messageBlockingQueue.poll(); + if (batchMessage != null){ + return getMessage(batchMessage); + } + return null; + } + + @Override + public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + try { + ConsumerBatchMessage batchMessage = messageBlockingQueue.poll(timeout, unit); + return getMessage(batchMessage); + } catch (InterruptedException ex) { + logger.warn("Get message timeout", ex); + throw new CanalClientException("failed to fetch the data after " + timeout); + } + } + + @Override + public void ack(long batchId) throws CanalClientException { + ConsumerBatchMessage batchMessage = messageCache.get(batchId); + batchMessage.ack(); + } + + @Override + public void rollback(long batchId) throws CanalClientException { + + } + + @Override + public void rollback() throws CanalClientException { + + } + + @Override + public void stopRunning() throws CanalClientException { + this.rocketMQConsumer.shutdown(); + connected = false; + } + +} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnectorProvider.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnectorProvider.java new file mode 100644 index 00000000..3b876f49 --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnectorProvider.java @@ -0,0 +1,20 @@ +package com.alibaba.otter.canal.client.rocketmq; + +/** + * RocketMQ connector provider. + */ +public class RocketMQCanalConnectorProvider { + /** + * Create RocketMQ connector + * + * @param nameServers name servers for RocketMQ + * @param topic + * @param groupId + * @return {@link RocketMQCanalConnector} + */ + public static RocketMQCanalConnector newRocketMQConnector(String nameServers, String topic, + String groupId) { + return new RocketMQCanalConnector(nameServers, topic, groupId); + } + +} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalListener.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalListener.java new file mode 100644 index 00000000..c30cd61f --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalListener.java @@ -0,0 +1,11 @@ +package com.alibaba.otter.canal.client.rocketmq; + +import java.util.List; +import org.apache.rocketmq.common.message.MessageExt; + +/** + * RocketMQ message listener + */ +public interface RocketMQCanalListener { + boolean onReceive(List messageExts); +} diff --git a/deployer/pom.xml b/deployer/pom.xml index 3580869b..690b8c91 100644 --- a/deployer/pom.xml +++ b/deployer/pom.xml @@ -40,7 +40,7 @@ **/canal.properties **/spring/** **/example/** - **/kafka.yml + **/mq.yml diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 2524186a..4388de47 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -1,24 +1,26 @@ package com.alibaba.otter.canal.deployer; +import com.alibaba.otter.canal.kafka.CanalKafkaProducer; +import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer; +import com.alibaba.otter.canal.server.CanalMQStarter; +import com.alibaba.otter.canal.spi.CanalMQProducer; import java.io.FileInputStream; import java.util.Properties; -import com.alibaba.otter.canal.kafka.CanalKafkaStarter; -import com.alibaba.otter.canal.server.CanalServerStarter; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * canal独立版本启动的入口类 - * + * * @author jianghang 2012-11-6 下午05:20:49 * @version 1.0.0 */ public class CanalLauncher { private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); + private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); public static void main(String[] args) throws Throwable { try { @@ -54,15 +56,18 @@ public class CanalLauncher { }); - CanalServerStarter canalServerStarter = null; + CanalMQProducer canalMQProducer = null; String serverMode = properties.getProperty(CanalConstants.CANAL_SERVER_MODE, "tcp"); if (serverMode.equalsIgnoreCase("kafka")) { - canalServerStarter = new CanalKafkaStarter(); - } else if (serverMode.equalsIgnoreCase("rocketMQ")) { - // 预留rocketMQ启动 + canalMQProducer = new CanalKafkaProducer(); + } else if (serverMode.equalsIgnoreCase("rocketmq")) { + canalMQProducer = new CanalRocketMQProducer(); } - if (canalServerStarter != null) { - canalServerStarter.init(); + if (canalMQProducer != null) { + CanalMQStarter canalServerStarter = new CanalMQStarter(canalMQProducer); + if (canalServerStarter != null) { + canalServerStarter.init(); + } } } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal Server:", e); diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 3d580bae..e2ad657b 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -9,8 +9,8 @@ canal.zkServers= # flush data to zk canal.zookeeper.flush.period = 1000 canal.withoutNetty = false -# tcp, kafka, rocketMQ -canal.serverMode = tcp +# tcp, kafka, RocketMQ +canal.serverMode = rocketmq # flush meta cursor/parse position to file canal.file.data.dir = ${canal.conf.dir} canal.file.flush.period = 1000 diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index f72b65fe..3cdbac4a 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -6,7 +6,7 @@ canal.instance.gtidon=false # position info -canal.instance.master.address=127.0.0.1:3306 +canal.instance.master.address=120.78.225.173:3307 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= @@ -18,8 +18,9 @@ canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info -canal.instance.tsdb.enable=true -#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb +#canal.instance.tsdb.enable=true +#canal.instance.tsdb.url=jdbc:mysql://120.78.225.173:3306/test +#canal.instance.mysql.slaveId = 1234 #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal @@ -31,7 +32,7 @@ canal.instance.tsdb.enable=true # username/password canal.instance.dbUsername=canal -canal.instance.dbPassword=canal +canal.instance.dbPassword=P@ssw0rd canal.instance.connectionCharset=UTF-8 # table regex diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/mq.yml similarity index 85% rename from deployer/src/main/resources/kafka.yml rename to deployer/src/main/resources/mq.yml index 93d4d557..d8b18f4a 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/mq.yml @@ -1,4 +1,4 @@ -servers: slave1:6667,slave2:6667,slave3:6667 +servers: localhost:9876 #for rocketmq: means the nameserver retries: 0 batchSize: 16384 lingerMs: 1 diff --git a/server/pom.xml b/server/pom.xml index a2eb437d..20108523 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -42,6 +42,11 @@ + + org.apache.rocketmq + rocketmq-client + 4.3.0 + org.jboss.netty diff --git a/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java new file mode 100644 index 00000000..55d350c6 --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java @@ -0,0 +1,70 @@ +package com.alibaba.otter.canal.common; + +import com.alibaba.otter.canal.protocol.CanalEntry; +import com.alibaba.otter.canal.protocol.CanalPacket; +import com.alibaba.otter.canal.protocol.Message; +import com.google.protobuf.ByteString; +import com.google.protobuf.CodedOutputStream; +import com.google.protobuf.WireFormat; +import java.util.List; +import org.apache.kafka.common.errors.SerializationException; +import org.springframework.util.CollectionUtils; + +public class CanalMessageSerializer { + public static byte[] serializer(Message data){ + try { + if (data != null) { + if (data.getId() != -1) { + if (data.isRaw() && !CollectionUtils.isEmpty(data.getRawEntries())) { + // for performance + List rowEntries = data.getRawEntries(); + // message size + int messageSize = 0; + messageSize += CodedOutputStream.computeInt64Size(1, data.getId()); + + int dataSize = 0; + for (int i = 0; i < rowEntries.size(); i++) { + dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i)); + } + messageSize += dataSize; + messageSize += 1 * rowEntries.size(); + // packet size + int size = 0; + size += CodedOutputStream.computeEnumSize(3, + CanalPacket.PacketType.MESSAGES.getNumber()); + size += CodedOutputStream.computeTagSize(5) + + CodedOutputStream.computeRawVarint32Size(messageSize) + + messageSize; + // build data + byte[] body = new byte[size]; + CodedOutputStream output = CodedOutputStream.newInstance(body); + output.writeEnum(3, CanalPacket.PacketType.MESSAGES.getNumber()); + + output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED); + output.writeRawVarint32(messageSize); + // message + output.writeInt64(1, data.getId()); + for (int i = 0; i < rowEntries.size(); i++) { + output.writeBytes(2, rowEntries.get(i)); + } + output.checkNoSpaceLeft(); + return body; + } else if (!CollectionUtils.isEmpty(data.getEntries())) { + CanalPacket.Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder(); + for (CanalEntry.Entry entry : data.getEntries()) { + messageBuilder.addMessages(entry.toByteString()); + } + + CanalPacket.Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder(); + packetBuilder.setType(CanalPacket.PacketType.MESSAGES); + packetBuilder.setBody(messageBuilder.build().toByteString()); + return packetBuilder.build().toByteArray(); + } + } + } + } catch (Exception e) { + throw new SerializationException("Error when serializing message to byte[] "); + } + return null; + } +} diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java similarity index 92% rename from server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java rename to server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index 140ce965..823a506a 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -1,4 +1,4 @@ -package com.alibaba.otter.canal.kafka; +package com.alibaba.otter.canal.common; import java.util.ArrayList; import java.util.HashSet; @@ -11,7 +11,7 @@ import java.util.Set; * @author machengyuan 2018-6-11 下午05:30:49 * @version 1.0.0 */ -public class KafkaProperties { +public class MQProperties { private String servers = "localhost:6667"; private int retries = 0; @@ -20,6 +20,7 @@ public class KafkaProperties { private long bufferMemory = 33554432L; private boolean filterTransactionEntry = true; private int canalBatchSize = 5; + private String producerGroup = "Canal-Producer"; private List canalDestinations = new ArrayList(); @@ -167,4 +168,11 @@ public class KafkaProperties { this.filterTransactionEntry = filterTransactionEntry; } + public String getProducerGroup() { + return producerGroup; + } + + public void setProducerGroup(String producerGroup) { + this.producerGroup = producerGroup; + } } diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index b6dba051..91531f7d 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.kafka; +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.spi.CanalMQProducer; import java.io.IOException; import java.util.Properties; @@ -18,13 +20,14 @@ import com.alibaba.otter.canal.protocol.Message; * @author machengyuan 2018-6-11 下午05:30:49 * @version 1.0.0 */ -public class CanalKafkaProducer { +public class CanalKafkaProducer implements CanalMQProducer { - private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); + private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); private Producer producer; - public void init(KafkaProperties kafkaProperties) { + @Override + public void init(MQProperties kafkaProperties) { Properties properties = new Properties(); properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("acks", "all"); @@ -37,6 +40,7 @@ public class CanalKafkaProducer { producer = new KafkaProducer(properties); } + @Override public void stop() { try { logger.info("## stop the kafka producer"); @@ -48,33 +52,8 @@ public class CanalKafkaProducer { } } - public void send(KafkaProperties.Topic topic, Message message) throws IOException { - // set canal.instance.filter.transaction.entry = true - - // boolean valid = false; - // if (message != null) { - // if (message.isRaw() && !message.getRawEntries().isEmpty()) { - // for (ByteString byteString : message.getRawEntries()) { - // CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString); - // if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN - // && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) { - // valid = true; - // break; - // } - // } - // } else if (!message.getEntries().isEmpty()){ - // for (CanalEntry.Entry entry : message.getEntries()) { - // if (entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN - // && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND) { - // valid = true; - // break; - // } - // } - // } - // } - // if (!valid) { - // return; - // } + @Override + public void send(MQProperties.Topic topic, Message message) throws IOException { ProducerRecord record; if (topic.getPartition() != null) { record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java index 49db5aa1..67dc04bc 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.kafka; +import com.alibaba.otter.canal.common.CanalMessageSerializer; import java.util.List; import java.util.Map; @@ -29,60 +30,7 @@ public class MessageSerializer implements Serializer { @Override public byte[] serialize(String topic, Message data) { - try { - if (data != null) { - if (data.getId() != -1) { - if (data.isRaw() && !CollectionUtils.isEmpty(data.getRawEntries())) { - // for performance - List rowEntries = data.getRawEntries(); - // message size - int messageSize = 0; - messageSize += CodedOutputStream.computeInt64Size(1, data.getId()); - - int dataSize = 0; - for (int i = 0; i < rowEntries.size(); i++) { - dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i)); - } - messageSize += dataSize; - messageSize += 1 * rowEntries.size(); - // packet size - int size = 0; - size += CodedOutputStream.computeEnumSize(3, - PacketType.MESSAGES.getNumber()); - size += CodedOutputStream.computeTagSize(5) - + CodedOutputStream.computeRawVarint32Size(messageSize) - + messageSize; - // build data - byte[] body = new byte[size]; - CodedOutputStream output = CodedOutputStream.newInstance(body); - output.writeEnum(3, PacketType.MESSAGES.getNumber()); - - output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED); - output.writeRawVarint32(messageSize); - // message - output.writeInt64(1, data.getId()); - for (int i = 0; i < rowEntries.size(); i++) { - output.writeBytes(2, rowEntries.get(i)); - } - output.checkNoSpaceLeft(); - return body; - } else if (!CollectionUtils.isEmpty(data.getEntries())) { - CanalPacket.Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder(); - for (CanalEntry.Entry entry : data.getEntries()) { - messageBuilder.addMessages(entry.toByteString()); - } - - CanalPacket.Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder(); - packetBuilder.setType(PacketType.MESSAGES); - packetBuilder.setBody(messageBuilder.build().toByteString()); - return packetBuilder.build().toByteArray(); - } - } - } - } catch (Exception e) { - throw new SerializationException("Error when serializing message to byte[] "); - } - return null; + return CanalMessageSerializer.serializer(data); } @Override 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 new file mode 100644 index 00000000..5a2900de --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -0,0 +1,58 @@ +package com.alibaba.otter.canal.rocketmq; + +import com.alibaba.otter.canal.common.CanalMessageSerializer; +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.server.exception.CanalServerException; +import com.alibaba.otter.canal.spi.CanalMQProducer; +import java.util.List; +import org.apache.rocketmq.client.exception.MQBrokerException; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.producer.DefaultMQProducer; +import org.apache.rocketmq.client.producer.MessageQueueSelector; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.remoting.exception.RemotingException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class CanalRocketMQProducer implements CanalMQProducer { + private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class); + + private DefaultMQProducer defaultMQProducer; + + @Override + public void init(MQProperties rocketMQProperties) { + defaultMQProducer = new DefaultMQProducer(); + defaultMQProducer.setNamesrvAddr(rocketMQProperties.getServers()); + defaultMQProducer.setProducerGroup(rocketMQProperties.getProducerGroup()); + defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProperties.getRetries()); + logger.info("##Start RocketMQ producer##"); + try { + defaultMQProducer.start(); + } catch (MQClientException ex) { + throw new CanalServerException("Start RocketMQ producer error", ex); + } + } + + @Override + public void send(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); + } + }, null); + } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { + logger.error("Send message error!", e); + } + } + + @Override + public void stop() { + logger.info("## Stop RocketMQ producer##"); + this.defaultMQProducer.shutdown(); + } +} diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java similarity index 66% rename from server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java rename to server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index d244ce31..bd2fdac2 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -1,69 +1,63 @@ -package com.alibaba.otter.canal.kafka; +package com.alibaba.otter.canal.server; +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.protocol.ClientIdentity; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; +import com.alibaba.otter.canal.spi.CanalMQProducer; import java.io.FileInputStream; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; - import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; -import com.alibaba.otter.canal.kafka.KafkaProperties.CanalDestination; -import com.alibaba.otter.canal.kafka.KafkaProperties.Topic; -import com.alibaba.otter.canal.protocol.ClientIdentity; -import com.alibaba.otter.canal.protocol.Message; -import com.alibaba.otter.canal.server.CanalServerStarter; -import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; - -/** - * kafka 启动类 - * - * @author machengyuan 2018-6-11 下午05:30:49 - * @version 1.0.0 - */ -public class CanalKafkaStarter implements CanalServerStarter { - - private static final Logger logger = LoggerFactory.getLogger(CanalKafkaStarter.class); +public class CanalMQStarter { + private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class); private static final String CLASSPATH_URL_PREFIX = "classpath:"; private volatile boolean running = false; - private ExecutorService executorService; + private ExecutorService executorService; - private CanalKafkaProducer canalKafkaProducer; + private CanalMQProducer canalMQProducer; - private KafkaProperties kafkaProperties; + private MQProperties properties; + + public CanalMQStarter(CanalMQProducer canalMQProducer){ + this.canalMQProducer = canalMQProducer; + } public void init() { try { - logger.info("## load kafka configurations"); - String conf = System.getProperty("kafka.conf", "classpath:kafka.yml"); + logger.info("## load MQ configurations"); + String conf = System.getProperty("mq.conf", "classpath:mq.yml"); if (conf.startsWith(CLASSPATH_URL_PREFIX)) { conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); - kafkaProperties = new Yaml().loadAs(CanalKafkaStarter.class.getClassLoader().getResourceAsStream(conf), - KafkaProperties.class); + properties = new Yaml().loadAs(CanalMQStarter.class.getClassLoader().getResourceAsStream(conf), + MQProperties.class); } else { - kafkaProperties = new Yaml().loadAs(new FileInputStream(conf), KafkaProperties.class); + properties = new Yaml().loadAs(new FileInputStream(conf), MQProperties.class); } // 初始化 kafka producer - canalKafkaProducer = new CanalKafkaProducer(); - canalKafkaProducer.init(kafkaProperties); +// canalMQProducer = new CanalKafkaProducer(); + canalMQProducer.init(properties); // set filterTransactionEntry - if (kafkaProperties.isFilterTransactionEntry()) { + if (properties.isFilterTransactionEntry()) { System.setProperty("canal.instance.filter.transaction.entry", "true"); } // 对应每个instance启动一个worker线程 - List destinations = kafkaProperties.getCanalDestinations(); + List destinations = properties.getCanalDestinations(); executorService = Executors.newFixedThreadPool(destinations.size()); - logger.info("## start the kafka workers."); - for (final CanalDestination destination : destinations) { + logger.info("## start the MQ workers."); + for (final MQProperties.CanalDestination destination : destinations) { executorService.execute(new Runnable() { @Override @@ -73,31 +67,31 @@ public class CanalKafkaStarter implements CanalServerStarter { }); } running = true; - logger.info("## the kafka workers is running now ......"); + logger.info("## the MQ workers is running now ......"); Runtime.getRuntime().addShutdownHook(new Thread() { public void run() { try { - logger.info("## stop the kafka workers"); + logger.info("## stop the MQ workers"); running = false; executorService.shutdown(); - canalKafkaProducer.stop(); + canalMQProducer.stop(); } catch (Throwable e) { - logger.warn("##something goes wrong when stopping kafka workers:", e); + logger.warn("##something goes wrong when stopping MQ workers:", e); } finally { - logger.info("## canal kafka is down."); + logger.info("## canal MQ is down."); } } }); } catch (Throwable e) { - logger.error("## Something goes wrong when starting up the canal kafka workers:", e); + logger.error("## Something goes wrong when starting up the canal MQ workers:", e); System.exit(0); } } - private void worker(CanalDestination destination) { + private void worker(MQProperties.CanalDestination destination) { while (!running) ; logger.info("## start the canal consumer: {}.", destination.getCanalDestination()); @@ -117,19 +111,19 @@ public class CanalKafkaStarter implements CanalServerStarter { logger.info("## the canal consumer {} is running now ......", destination.getCanalDestination()); while (running) { - Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据 + Message message = server.getWithoutAck(clientIdentity, properties.getCanalBatchSize()); // 获取指定数量的数据 long batchId = message.getId(); try { int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); if (batchId != -1 && size != 0) { if (!StringUtils.isEmpty(destination.getTopic())) { - Topic topic = new Topic(); + MQProperties.Topic topic = new MQProperties.Topic(); topic.setTopic(destination.getTopic()); topic.setPartition(destination.getPartition()); destination.getTopics().add(topic); } - for (Topic topic : destination.getTopics()) { - canalKafkaProducer.send(topic, message); // 发送message到所有topic + for (MQProperties.Topic topic : destination.getTopics()) { + canalMQProducer.send(topic, message); // 发送message到所有topic } } diff --git a/server/src/main/java/com/alibaba/otter/canal/spi/CanalMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMQProducer.java new file mode 100644 index 00000000..b0edaf5b --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/spi/CanalMQProducer.java @@ -0,0 +1,28 @@ +package com.alibaba.otter.canal.spi; + +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.protocol.Message; +import java.io.IOException; + +public interface CanalMQProducer { + /** + * Init producer. + * + * @param mqProperties MQ config + */ + void init(MQProperties mqProperties); + + /** + * Send canal message to related topic + * + * @param topic MQ topic + * @param message canal message + * @throws IOException + */ + void send(MQProperties.Topic topic, Message message) throws IOException; + + /** + * Stop MQ producer service + */ + void stop(); +} \ No newline at end of file From 0cda90cdeae57c4d30468ffa581e105c784695da Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 16:17:38 +0800 Subject: [PATCH 003/328] =?UTF-8?q?=E9=BB=98=E8=AE=A4=E5=B0=86=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=8F=91=E7=8E=B0kafka=E7=9A=84partition=200=EF=BC=8C?= =?UTF-8?q?=E9=81=BF=E5=85=8D=E5=A4=9A=E5=88=86=E5=8C=BA=E7=9A=84=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=B8=80=E8=87=B4=E6=80=A7=E9=97=AE=E9=A2=98=20?= =?UTF-8?q?=E5=90=8E=E7=BB=AD=E5=B0=86=E6=94=AF=E6=8C=81=E9=80=9A=E8=BF=87?= =?UTF-8?q?id=20hash=E5=88=86=E5=8F=91=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/kafka/CanalKafkaProducer.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 4942a37b..e1ea5e47 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -75,7 +75,7 @@ public class CanalKafkaProducer { if (topic.getPartition() != null) { record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); } else { - record = new ProducerRecord(topic.getTopic(), message); + record = new ProducerRecord(topic.getTopic(), 0, null, message); } producer.send(record); @@ -84,8 +84,18 @@ public class CanalKafkaProducer { List flatMessages = FlatMessage.messageConverter(message); if (flatMessages != null) { for (FlatMessage flatMessage : flatMessages) { - ProducerRecord record = new ProducerRecord(topic.getTopic(), - JSON.toJSONString(flatMessage)); + ProducerRecord record; + if (topic.getPartition() != null) { + record = new ProducerRecord(topic.getTopic(), + topic.getPartition(), + null, + JSON.toJSONString(flatMessage)); + } else { + record = new ProducerRecord(topic.getTopic(), + 0, + null, + JSON.toJSONString(flatMessage)); + } producer2.send(record); } } From 74890050d0c323877eb9c1e3dec128e989eb6ee8 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 16:48:14 +0800 Subject: [PATCH 004/328] =?UTF-8?q?kafka=E7=94=9F=E4=BA=A7=E7=AB=AF?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=8C=89pk=20hash=E5=88=B0=E5=AF=B9=E5=BA=94?= =?UTF-8?q?partition=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/kafka.yml | 6 +- .../otter/canal/protocol/FlatMessage.java | 31 ++++++---- .../otter/canal/kafka/CanalKafkaProducer.java | 55 ++++++++++++----- .../otter/canal/kafka/CanalKafkaStarter.java | 5 +- .../otter/canal/kafka/KafkaProperties.java | 60 +++++-------------- 5 files changed, 79 insertions(+), 78 deletions(-) diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 735f723d..b86b3034 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -12,7 +12,9 @@ flatMessage: true canalDestinations: - canalDestination: example - topic: example - partition: + topic: expample + #partitionsNum: 3 + #partitionHash: + # mytest.person: id diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java index d7150b79..51bd320a 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -1,14 +1,8 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; -import com.google.common.collect.Table; import com.google.protobuf.ByteString; /** @@ -31,7 +25,7 @@ public class FlatMessage implements Serializable { private List> data; private List> old; - public FlatMessage() { + public FlatMessage(){ } public FlatMessage(long id){ @@ -126,6 +120,12 @@ public class FlatMessage implements Serializable { this.old = old; } + /** + * 将Message转换为FlatMessage + * + * @param message 原生message + * @return FlatMessage列表 + */ public static List messageConverter(Message message) { try { if (message == null) { @@ -231,11 +231,22 @@ public class FlatMessage implements Serializable { } } + /** + * 将FlatMessage按指定的字段值hash拆分 + * + * @param flatMessage flatMessage + * @param partitionsNum 分区数量 + * @param pkHashConfig hash映射 + * @return 拆分后的flatMessage数组 + */ public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, - Table pkHashConfig) { + Map pkHashConfig) { + if (partitionsNum == null) { + partitionsNum = 1; + } FlatMessage[] partitionMessages = new FlatMessage[partitionsNum]; - String pk = pkHashConfig.get(flatMessage.getDatabase(), flatMessage.getTable()); + String pk = pkHashConfig.get(flatMessage.getDatabase() + "." + flatMessage.getTable()); if (pk == null || flatMessage.getIsDdl()) { partitionMessages[0] = flatMessage; } else { diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index e1ea5e47..fbeeeb87 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -67,15 +67,18 @@ public class CanalKafkaProducer { } } - public void send(KafkaProperties.Topic topic, Message message, Callback callback) { + public void send(KafkaProperties.CanalDestination canalDestination, Message message, Callback callback) { try { // producer.beginTransaction(); if (!kafkaProperties.getFlatMessage()) { ProducerRecord record; - if (topic.getPartition() != null) { - record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); + if (canalDestination.getPartition() != null) { + record = new ProducerRecord(canalDestination.getTopic(), + canalDestination.getPartition(), + null, + message); } else { - record = new ProducerRecord(topic.getTopic(), 0, null, message); + record = new ProducerRecord(canalDestination.getTopic(), 0, null, message); } producer.send(record); @@ -84,19 +87,39 @@ public class CanalKafkaProducer { List flatMessages = FlatMessage.messageConverter(message); if (flatMessages != null) { for (FlatMessage flatMessage : flatMessages) { - ProducerRecord record; - if (topic.getPartition() != null) { - record = new ProducerRecord(topic.getTopic(), - topic.getPartition(), - null, - JSON.toJSONString(flatMessage)); + if (canalDestination.getPartition() != null) { + ProducerRecord record = new ProducerRecord(canalDestination + .getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage)); + producer2.send(record); } else { - record = new ProducerRecord(topic.getTopic(), - 0, - null, - JSON.toJSONString(flatMessage)); + if (canalDestination.getPartitionHash() != null + && !canalDestination.getPartitionHash().isEmpty()) { + FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage, + canalDestination.getPartitionsNum(), + canalDestination.getPartitionHash()); + int length = partitionFlatMessage.length; + for (int i = 0; i < length; i++) { + FlatMessage flatMessage1 = partitionFlatMessage[i]; + if (flatMessage1 == null) { + continue; + } + ProducerRecord record = new ProducerRecord( + canalDestination.getTopic(), + i, + null, + JSON.toJSONString(partitionFlatMessage[i])); + producer2.send(record); + } + } else { + ProducerRecord record = new ProducerRecord( + canalDestination.getTopic(), + 0, + null, + JSON.toJSONString(flatMessage)); + producer2.send(record); + } } - producer2.send(record); + } } } @@ -104,7 +127,7 @@ public class CanalKafkaProducer { // producer.commitTransaction(); callback.commit(); if (logger.isDebugEnabled()) { - logger.debug("send message to kafka topic: {}", topic.getTopic()); + logger.debug("send message to kafka topic: {}", canalDestination.getTopic()); } } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index 5b73bf3b..abe7bf09 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -132,10 +132,7 @@ public class CanalKafkaStarter implements CanalServerStarter { try { int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); if (batchId != -1 && size != 0) { - Topic topic = new Topic(); - topic.setTopic(destination.getTopic()); - topic.setPartition(destination.getPartition()); - canalKafkaProducer.send(topic, message, new CanalKafkaProducer.Callback() { + canalKafkaProducer.send(destination, message, new CanalKafkaProducer.Callback() { @Override public void commit() { diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java index 82707ba4..5e4145fe 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java @@ -1,9 +1,8 @@ package com.alibaba.otter.canal.kafka; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; -import java.util.Set; +import java.util.Map; /** * kafka 配置项 @@ -27,10 +26,11 @@ public class KafkaProperties { public static class CanalDestination { - private String canalDestination; - private String topic; - private Integer partition; - private Set topics = new HashSet(); + private String canalDestination; + private String topic; + private Integer partition; + private Integer partitionsNum; + private Map partitionHash; public String getCanalDestination() { return canalDestination; @@ -56,52 +56,20 @@ public class KafkaProperties { this.partition = partition; } - public Set getTopics() { - return topics; + public Integer getPartitionsNum() { + return partitionsNum; } - public void setTopics(Set topics) { - this.topics = topics; - } - } - - public static class Topic { - - private String topic; - private Integer partition; - - public String getTopic() { - return topic; + public void setPartitionsNum(Integer partitionsNum) { + this.partitionsNum = partitionsNum; } - public void setTopic(String topic) { - this.topic = topic; + public Map getPartitionHash() { + return partitionHash; } - public Integer getPartition() { - return partition; - } - - public void setPartition(Integer partition) { - this.partition = partition; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - - Topic topic1 = (Topic) o; - - if (topic != null ? !topic.equals(topic1.topic) : topic1.topic != null) return false; - return partition != null ? partition.equals(topic1.partition) : topic1.partition == null; - } - - @Override - public int hashCode() { - int result = topic != null ? topic.hashCode() : 0; - result = 31 * result + (partition != null ? partition.hashCode() : 0); - return result; + public void setPartitionHash(Map partitionHash) { + this.partitionHash = partitionHash; } } From e566160f1ea24c3db15b45f51b1b6ad2ce226b42 Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 17:24:32 +0800 Subject: [PATCH 005/328] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/kafka.yml | 10 ++++++---- .../otter/canal/kafka/CanalKafkaProducer.java | 17 ++++++++--------- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index b86b3034..76e2bade 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -12,9 +12,11 @@ flatMessage: true canalDestinations: - canalDestination: example - topic: expample - #partitionsNum: 3 - #partitionHash: - # mytest.person: id + topic: example +# #对应topic分区数量 +# partitionsNum: 3 +# partitionHash: +# #库名.表名: 唯一主键 +# mytest.person: id diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index fbeeeb87..5708105c 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -99,16 +99,15 @@ public class CanalKafkaProducer { canalDestination.getPartitionHash()); int length = partitionFlatMessage.length; for (int i = 0; i < length; i++) { - FlatMessage flatMessage1 = partitionFlatMessage[i]; - if (flatMessage1 == null) { - continue; + FlatMessage flatMessagePart = partitionFlatMessage[i]; + if (flatMessagePart != null) { + ProducerRecord record = new ProducerRecord( + canalDestination.getTopic(), + i, + null, + JSON.toJSONString(flatMessagePart)); + producer2.send(record); } - ProducerRecord record = new ProducerRecord( - canalDestination.getTopic(), - i, - null, - JSON.toJSONString(partitionFlatMessage[i])); - producer2.send(record); } } else { ProducerRecord record = new ProducerRecord( From 4f2b4cea53e9dfc526f6600fbf1c0a3bf955dabd Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 17:25:32 +0800 Subject: [PATCH 006/328] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deployer/src/main/resources/kafka.yml | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 76e2bade..aa23c1ee 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -11,12 +11,12 @@ canalGetTimeout: 100 flatMessage: true canalDestinations: - - canalDestination: example - topic: example -# #对应topic分区数量 -# partitionsNum: 3 -# partitionHash: -# #库名.表名: 唯一主键 -# mytest.person: id +- canalDestination: example + topic: exp3 +# #对应topic分区数量 +# partitionsNum: 3 +# partitionHash: +# #库名.表名: 唯一主键 +# mytest.person: id From 19b7f730b640471150796f487f002378dca0cc0b Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 20 Sep 2018 18:06:47 +0800 Subject: [PATCH 007/328] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java | 1 - 1 file changed, 1 deletion(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index abe7bf09..f2df3547 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -12,7 +12,6 @@ import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; import com.alibaba.otter.canal.kafka.KafkaProperties.CanalDestination; -import com.alibaba.otter.canal.kafka.KafkaProperties.Topic; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.server.CanalServerStarter; From fd80f953e97eaf741ace487f950c8d2492ea60a8 Mon Sep 17 00:00:00 2001 From: agapple Date: Fri, 21 Sep 2018 15:30:35 +0800 Subject: [PATCH 008/328] Update README.md --- README.md | 44 +++++++++++++++++++++++++++++--------------- 1 file changed, 29 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 86ef8755..2d7bc589 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,7 @@
  1. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
  2. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
    3. +
  3. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
  4. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
    5. DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>>
@@ -52,18 +53,38 @@ See the wiki page for : wiki文

wiki文档列表

+

多语言业务使用

+ +1. canal整体交互协议设计上使用了protobuf3.0，理论上可以支持绝大部分的多语言场景，欢迎大家提交多客户端的PR + * canal java客户端: https://github.com/alibaba/canal/wiki/ClientExample + * canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp + * canal go客户端，开发进行中 +2. canal作为MySQL binlog的增量获取工具，可以将数据投递到MQ系统中，比如Kafka/RocketMQ，可以借助于MQ的多语言能力 +

相关资料

* ADC阿里技术嘉年华分享ppt (放在google docs上，可能需要翻墙): ppt下载 @@ -90,10 +111,3 @@ See the wiki page for : wiki文
  • 新浪微博： agapple0002
  • 报告issue：issues
    • - -
    -【招聘】阿里巴巴中间件团队招聘JAVA高级工程师
-岗位主要为技术型内容(非业务部门)，阿里中间件整个体系对于未来想在技术上有所沉淀的同学还是非常有帮助的
-工作地点：杭州、北京均可. ps. 阿里待遇向来都是不错的，有意者可以QQ、微博私聊. 
-具体招聘内容：https://job.alibaba.com/zhaopin/position_detail.htm?positionId=32666
-
    From 9535e82da6a3ed760919747122f25b8ee1ae334e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 21 Sep 2018 15:40:18 +0800 Subject: [PATCH 009/328] fixed readme --- README.md | 17 +++++++++++++++-- .../resources/spring/tsdb/sql/create_table.sql | 8 ++++---- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 2d7bc589..95331c55 100644 --- a/README.md +++ b/README.md @@ -3,9 +3,11 @@

    最新更新

      +
    1. canal发布重大版本更新1.1.0，具体releaseNode参考：https://github.com/alibaba/canal/releases/tag/canal-1.1.0
      2. +
    2. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
    3. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
    4. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
      5. -
    5. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
      6. +
    6. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
      7. DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>>
    @@ -47,6 +49,16 @@ DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体
  • canal解析binary log对象(原始为byte流)
    • +

    重要版本更新说明

    + +canal 1.1.x系列，参考release文档：版本发布信息： + +1. 整体性能测试&优化,提升了150%. #726 参考: 【[Performance](https://github.com/alibaba/canal/wiki/Performance)】 +2. 原生支持prometheus监控 #765 【[Prometheus QuickStart](https://github.com/alibaba/canal/wiki/Prometheus-QuickStart)】 +3. 原生支持kafka消息投递 #695 【[Canal Kafka QuickStart](https://github.com/alibaba/canal/wiki/Canal-Kafka-QuickStart)】 +4. 原生支持aliyun rds的binlog订阅 (解决自动主备切换/oss binlog离线解析) 参考: 【[Aliyun RDS QuickStart](https://github.com/alibaba/canal/wiki/aliyun-RDS-QuickStart)】 +5. 原生支持docker镜像 #801 参考: 【[Docker QuickStart](https://github.com/alibaba/canal/wiki/Docker-QuickStart)】 +

    相关文档

    See the wiki page for : wiki文档 @@ -77,7 +89,7 @@ See the wiki page for : wiki文
  • FAQ
    • -

    多语言业务使用

    +

    多语言业务

    1. canal整体交互协议设计上使用了protobuf3.0，理论上可以支持绝大部分的多语言场景，欢迎大家提交多客户端的PR * canal java客户端: https://github.com/alibaba/canal/wiki/ClientExample @@ -111,3 +123,4 @@ See the wiki page for : wiki文
  • 新浪微博： agapple0002
  • 报告issue：issues
    • + diff --git a/deployer/src/main/resources/spring/tsdb/sql/create_table.sql b/deployer/src/main/resources/spring/tsdb/sql/create_table.sql index e5d6dbf4..322bc2b6 100644 --- a/deployer/src/main/resources/spring/tsdb/sql/create_table.sql +++ b/deployer/src/main/resources/spring/tsdb/sql/create_table.sql @@ -26,10 +26,10 @@ CREATE TABLE IF NOT EXISTS `meta_history` ( `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema', - `schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema', - `table` varchar(1024) DEFAULT NULL COMMENT '对应的table', - `sql` longtext DEFAULT NULL COMMENT '执行的sql', - `type` varchar(256) DEFAULT NULL COMMENT 'sql类型', + `sql_schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema', + `sql_table` varchar(1024) DEFAULT NULL COMMENT '对应的table', + `sql_text` longtext DEFAULT NULL COMMENT '执行的sql', + `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql类型', `extra` text DEFAULT NULL COMMENT '额外的扩展信息', PRIMARY KEY (`id`), UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), From 9d48d23787c8e24689c7a3c8b431c6109e0a5d4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 21 Sep 2018 15:44:03 +0800 Subject: [PATCH 010/328] fixed readme --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 95331c55..9a87de8f 100644 --- a/README.md +++ b/README.md @@ -51,7 +51,7 @@ DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体

    重要版本更新说明

    -canal 1.1.x系列，参考release文档：版本发布信息： +canal 1.1.x系列，参考release文档：版本发布信息 1. 整体性能测试&优化,提升了150%. #726 参考: 【[Performance](https://github.com/alibaba/canal/wiki/Performance)】 2. 原生支持prometheus监控 #765 【[Prometheus QuickStart](https://github.com/alibaba/canal/wiki/Prometheus-QuickStart)】 From 00ec70d425e5ebbd0dbcb7211c7354325af99be6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 21 Sep 2018 15:45:52 +0800 Subject: [PATCH 011/328] fixed readme --- README.md | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 9a87de8f..c28fc0ea 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,6 @@
    - -

    最新更新

    -
      -
    1. canal发布重大版本更新1.1.0，具体releaseNode参考：https://github.com/alibaba/canal/releases/tag/canal-1.1.0
      2. -
    2. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
      3. -
    3. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
      4. -
    4. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
      5. - -
    5. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
      6. -DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>> -
    - +

    背景

    早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务，从此开启了一段新纪元。

    ps. 目前内部版本已经支持mysql和oracle部分版本的日志解析，当前的canal开源版本支持5.7及以下的版本(阿里内部mysql 5.7.13, 5.6.10, mysql 5.5.18和5.1.40/48)

    @@ -124,3 +113,13 @@ See the wiki page for : wiki文
  • 报告issue：issues
    • +

    最新更新

    +
      +
    1. canal发布重大版本更新1.1.0，具体releaseNode参考：https://github.com/alibaba/canal/releases/tag/canal-1.1.0
      2. +
    2. canal c#客户端开源项目地址： https://github.com/CanalSharp/CanalSharp ，推荐!
      3. +
    3. canal QQ讨论群已经建立，群号：161559791 ，欢迎加入进行技术讨论。
      4. +
    4. canal消费端项目开源: Otter(分布式数据库同步系统)，地址：https://github.com/alibaba/otter
      5. + +
    5. Canal已在阿里云推出商业化版本 数据传输服务DTS， 开通即用，免去部署维护的昂贵使用成本。DTS针对阿里云RDS、DRDS等产品进行了适配，解决了Binlog日志回收，主备切换、VPC网络切换等场景下的订阅高可用问题。同时，针对RDS进行了针对性的性能优化。出于稳定性、性能及成本的考虑，强烈推荐阿里云用户使用DTS产品。DTS产品使用文档
      6. +DTS支持阿里云RDS&DRDS的Binlog日志实时订阅，现推出首月免费体验，限时限量，立即体验>>> +
    \ No newline at end of file From 821f3eced54b8a102fbc25f6ff92946f34fe426e Mon Sep 17 00:00:00 2001 From: duhengforever Date: Fri, 21 Sep 2018 18:41:37 +0800 Subject: [PATCH 012/328] 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) { From a22517adc21101d85c56acb7a839c9bb5e92d1d5 Mon Sep 17 00:00:00 2001 From: XiaoFu Date: Tue, 25 Sep 2018 17:53:54 +0800 Subject: [PATCH 013/328] Fix link error --- README.md | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index c28fc0ea..9fde64b5 100644 --- a/README.md +++ b/README.md @@ -54,25 +54,25 @@ See the wiki page for : wiki文

    wiki文档列表