From 50630cc9344d966cb4daaa3ca33d6d2b8abbbdc6 Mon Sep 17 00:00:00 2001 From: "maowei.ymw" Date: Mon, 19 Nov 2018 14:33:59 +0800 Subject: [PATCH 1/5] Support client connect rocketmq in aliyun --- client/pom.xml | 8 ++++++ .../rocketmq/RocketMQCanalConnector.java | 26 ++++++++++++++++++- pom.xml | 10 +++++++ server/pom.xml | 5 +++- .../otter/canal/common/MQProperties.java | 17 ++++++++++++ .../canal/rocketmq/CanalRocketMQProducer.java | 16 ++++++++++-- 6 files changed, 78 insertions(+), 4 deletions(-) diff --git a/client/pom.xml b/client/pom.xml index aa31390a..5994c860 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -122,6 +122,14 @@ junit test + + com.aliyun.openservices + aliware-apache-rocketmq-cloud + + + org.apache.rocketmq + rocketmq-client + 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 index da21ead8..bdb654de 100644 --- 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 @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.client.rocketmq; +import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials; +import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; @@ -10,8 +12,10 @@ 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.consumer.rebalance.AllocateMessageQueueAveragely; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.remoting.RPCHook; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -47,6 +51,9 @@ public class RocketMQCanalConnector implements CanalMQConnector { private long batchProcessTimeout = 60 * 1000; private boolean flatMessage; private volatile ConsumerBatchMessage lastGetBatchMessage = null; + private String accessKey; + private String secretKey; + public RocketMQCanalConnector(String nameServer, String topic, String groupName, boolean flatMessage){ this.nameServer = nameServer; @@ -56,8 +63,25 @@ public class RocketMQCanalConnector implements CanalMQConnector { this.messageBlockingQueue = new LinkedBlockingQueue<>(1024); } + public RocketMQCanalConnector(String nameServer, String topic, String groupName, + String accessKey, String secretKey, boolean flatMessage){ + this(nameServer, topic, groupName, flatMessage); + this.accessKey = accessKey; + this.secretKey = secretKey; + } + public void connect() throws CanalClientException { - rocketMQConsumer = new DefaultMQPushConsumer(groupName); + + RPCHook rpcHook = null; + if(null != accessKey && accessKey.length() > 0 + && null != secretKey && secretKey.length() > 0){ + SessionCredentials sessionCredentials = new SessionCredentials(); + sessionCredentials.setAccessKey(accessKey); + sessionCredentials.setSecretKey(secretKey); + rpcHook = new ClientRPCHook(sessionCredentials); + } + rocketMQConsumer = new DefaultMQPushConsumer(groupName, rpcHook, new AllocateMessageQueueAveragely()); + rocketMQConsumer.setVipChannelEnabled(false); if (!StringUtils.isBlank(nameServer)) { rocketMQConsumer.setNamesrvAddr(nameServer); } diff --git a/pom.xml b/pom.xml index 0bb4bc11..2c2e8c8e 100644 --- a/pom.xml +++ b/pom.xml @@ -302,6 +302,16 @@ jsr305 3.0.2 + + com.aliyun.openservices + aliware-apache-rocketmq-cloud + 1.0-SNAPSHOT + + + org.apache.rocketmq + rocketmq-client + 4.3.0 + diff --git a/server/pom.xml b/server/pom.xml index 3d3084e7..fd1225f4 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -40,7 +40,6 @@ org.apache.rocketmq rocketmq-client - 4.3.0 @@ -55,5 +54,9 @@ junit test + + com.aliyun.openservices + aliware-apache-rocketmq-cloud + diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index 3bc3394b..2e6fbe69 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -22,6 +22,8 @@ public class MQProperties { private boolean flatMessage = true; private String compressionType = "none"; private String acks = "all"; + private String aliyunAccessKey = ""; + private String aliyunSecretKey = ""; public static class CanalDestination { @@ -168,4 +170,19 @@ public class MQProperties { this.acks = acks; } + public String getAliyunAccessKey() { + return aliyunAccessKey; + } + + public void setAliyunAccessKey(String aliyunAccessKey) { + this.aliyunAccessKey = aliyunAccessKey; + } + + public String getAliyunSecretKey() { + return aliyunSecretKey; + } + + public void setAliyunSecretKey(String aliyunSecretKey) { + this.aliyunSecretKey = aliyunSecretKey; + } } 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 3501930b..74e3920b 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 @@ -7,12 +7,15 @@ import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.server.exception.CanalServerException; import com.alibaba.otter.canal.spi.CanalMQProducer; +import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials; +import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook; 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.RPCHook; import org.apache.rocketmq.remoting.exception.RemotingException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,10 +33,19 @@ public class CanalRocketMQProducer implements CanalMQProducer { @Override public void init(MQProperties rocketMQProperties) { this.mqProperties = rocketMQProperties; - defaultMQProducer = new DefaultMQProducer(); + RPCHook rpcHook = null; + if(rocketMQProperties.getAliyunAccessKey().length() > 0 + && rocketMQProperties.getAliyunSecretKey().length() > 0){ + SessionCredentials sessionCredentials = new SessionCredentials(); + sessionCredentials.setAccessKey(rocketMQProperties.getAliyunAccessKey()); + sessionCredentials.setSecretKey(rocketMQProperties.getAliyunSecretKey()); + rpcHook = new ClientRPCHook(sessionCredentials); + } + + defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(), rpcHook); defaultMQProducer.setNamesrvAddr(rocketMQProperties.getServers()); - defaultMQProducer.setProducerGroup(rocketMQProperties.getProducerGroup()); defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProperties.getRetries()); + defaultMQProducer.setVipChannelEnabled(false); logger.info("##Start RocketMQ producer##"); try { defaultMQProducer.start(); From 4bfc476e08b6ab41f33921ab0a25ef1b1eb1ebc4 Mon Sep 17 00:00:00 2001 From: "maowei.ymw" Date: Wed, 21 Nov 2018 10:44:58 +0800 Subject: [PATCH 2/5] Modify rocketMQ related pom --- client/pom.xml | 7 ------- pom.xml | 2 +- 2 files changed, 1 insertion(+), 8 deletions(-) diff --git a/client/pom.xml b/client/pom.xml index 5994c860..28ca8d27 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -101,12 +101,9 @@ ${spring_version} test - org.apache.rocketmq rocketmq-client - 4.3.0 - provided @@ -126,10 +123,6 @@ com.aliyun.openservices aliware-apache-rocketmq-cloud - - org.apache.rocketmq - rocketmq-client - diff --git a/pom.xml b/pom.xml index 2c2e8c8e..92d8a530 100644 --- a/pom.xml +++ b/pom.xml @@ -305,7 +305,7 @@ com.aliyun.openservices aliware-apache-rocketmq-cloud - 1.0-SNAPSHOT + 1.0 org.apache.rocketmq From 638952fad19768cb748f666cee0e311fe598c826 Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 21 Nov 2018 13:51:34 +0800 Subject: [PATCH 3/5] fixed binlog seek & aliyun ak/sk --- .../taobao/tddl/dbsync/binlog/LogDecoder.java | 11 +- .../otter/canal/deployer/CanalConstants.java | 2 + .../otter/canal/deployer/CanalController.java | 156 +++++++++--------- .../otter/canal/deployer/CanalLauncher.java | 9 + deployer/src/main/resources/canal.properties | 10 +- .../main/resources/spring/base-instance.xml | 4 +- .../parse/inbound/mysql/MysqlEventParser.java | 2 +- .../inbound/mysql/dbsync/LogEventConvert.java | 2 +- .../otter/canal/common/MQProperties.java | 8 +- 9 files changed, 111 insertions(+), 93 deletions(-) diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java index 41739cf8..3c1bff17 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java @@ -109,11 +109,6 @@ public final class LogDecoder { try { /* Decoding binary-log to event */ event = decode(buffer, header, context); - if (event != null) { - // set logFileName - event.getHeader().setLogFileName(context.getLogPosition().getFileName()); - event.setSemival(buffer.semival); - } } catch (IOException e) { if (logger.isWarnEnabled()) { logger.warn("Decoding " + LogEvent.getTypeName(header.getType()) + " failed from: " @@ -128,6 +123,12 @@ public final class LogDecoder { event = new UnknownLogEvent(header); } + if (event != null) { + // set logFileName + event.getHeader().setLogFileName(context.getLogPosition().getFileName()); + event.setSemival(buffer.semival); + } + /* consume this binary-log. */ buffer.consume(len); return event; diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index aa9b2909..d21348bf 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -47,6 +47,8 @@ public class CanalConstants { public static final String CANAL_MQ_FLATMESSAGE = ROOT + "." + "mq.flatMessage"; public static final String CANAL_MQ_COMPRESSION_TYPE = ROOT + "." + "mq.compressionType"; public static final String CANAL_MQ_ACKS = ROOT + "." + "mq.acks"; + public static final String CANAL_ALIYUN_ACCESSKEY = ROOT + "." + "aliyun.accessKey"; + public static final String CANAL_ALIYUN_SECRETKEY = ROOT + "." + "aliyun.secretKey"; public static String getInstanceModeKey(String destination) { return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index 9161bff0..dbc5938e 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -17,7 +17,6 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import com.alibaba.otter.canal.common.CanalException; -import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.common.zookeeper.ZkClientx; import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils; @@ -96,6 +95,16 @@ public class CanalController { System.setProperty(CanalConstants.CANAL_SOCKETCHANNEL, socketChannel); } + // 兼容1.1.0版本的ak/sk参数名 + String accesskey = getProperty(properties, "canal.instance.rds.accesskey"); + String secretkey = getProperty(properties, "canal.instance.rds.secretkey"); + if (StringUtils.isNotEmpty(accesskey)) { + System.setProperty(CanalConstants.CANAL_ALIYUN_ACCESSKEY, accesskey); + } + if (StringUtils.isNotEmpty(secretkey)) { + System.setProperty(CanalConstants.CANAL_ALIYUN_SECRETKEY, secretkey); + } + // 准备canal server cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID)); ip = getProperty(properties, CanalConstants.CANAL_IP); @@ -131,81 +140,80 @@ public class CanalController { final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port); ServerRunningMonitors.setServerData(serverData); - ServerRunningMonitors - .setRunningMonitors(MigrateMap.makeComputingMap(new Function() { + ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function() { - public ServerRunningMonitor apply(final String destination) { - ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData); - runningMonitor.setDestination(destination); - runningMonitor.setListener(new ServerRunningListener() { + public ServerRunningMonitor apply(final String destination) { + ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData); + runningMonitor.setDestination(destination); + runningMonitor.setListener(new ServerRunningListener() { - public void processActiveEnter() { - try { - MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); - embededCanalServer.start(destination); - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); - } + public void processActiveEnter() { + try { + MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); + embededCanalServer.start(destination); + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); } - - public void processActiveExit() { - try { - MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); - embededCanalServer.stop(destination); - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); - } - } - - public void processStart() { - try { - if (zkclientx != null) { - final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, - ip + ":" + port); - initCid(path); - zkclientx.subscribeStateChanges(new IZkStateListener() { - - public void handleStateChanged(KeeperState state) throws Exception { - - } - - public void handleNewSession() throws Exception { - initCid(path); - } - - @Override - public void handleSessionEstablishmentError(Throwable error) throws Exception { - logger.error("failed to connect to zookeeper", error); - } - }); - } - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); - } - } - - public void processStop() { - try { - MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); - if (zkclientx != null) { - final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, - ip + ":" + port); - releaseCid(path); - } - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); - } - } - - }); - if (zkclientx != null) { - runningMonitor.setZkClient(zkclientx); } - // 触发创建一下cid节点 - runningMonitor.init(); - return runningMonitor; + + public void processActiveExit() { + try { + MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); + embededCanalServer.stop(destination); + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); + } + } + + public void processStart() { + try { + if (zkclientx != null) { + final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" + + port); + initCid(path); + zkclientx.subscribeStateChanges(new IZkStateListener() { + + public void handleStateChanged(KeeperState state) throws Exception { + + } + + public void handleNewSession() throws Exception { + initCid(path); + } + + @Override + public void handleSessionEstablishmentError(Throwable error) throws Exception { + logger.error("failed to connect to zookeeper", error); + } + }); + } + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); + } + } + + public void processStop() { + try { + MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); + if (zkclientx != null) { + final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" + + port); + releaseCid(path); + } + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); + } + } + + }); + if (zkclientx != null) { + runningMonitor.setZkClient(zkclientx); } - })); + // 触发创建一下cid节点 + runningMonitor.init(); + return runningMonitor; + } + })); // 初始化monitor机制 autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN)); @@ -257,8 +265,7 @@ public class CanalController { instanceConfigMonitors = MigrateMap.makeComputingMap(new Function() { public InstanceConfigMonitor apply(InstanceMode mode) { - int scanInterval = Integer - .valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL)); + int scanInterval = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL)); if (mode.isSpring()) { SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor(); @@ -366,8 +373,7 @@ public class CanalController { InstanceConfig oldConfig = instanceConfigs.put(destination, config); if (oldConfig != null) { - logger - .warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config); + logger.warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config); } } } 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 80224380..24d0ec5b 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 @@ -125,6 +125,15 @@ public class CanalLauncher { if (!StringUtils.isEmpty(acks)) { mqProperties.setAcks(acks); } + + String aliyunAccessKey = CanalController.getProperty(properties, CanalConstants.CANAL_ALIYUN_ACCESSKEY); + if (!StringUtils.isEmpty(aliyunAccessKey)) { + mqProperties.setAliyunAccessKey(aliyunAccessKey); + } + String aliyunSecretKey = CanalController.getProperty(properties, CanalConstants.CANAL_ALIYUN_SECRETKEY); + if (!StringUtils.isEmpty(aliyunSecretKey)) { + mqProperties.setAliyunSecretKey(aliyunSecretKey); + } return mqProperties; } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index d555c162..763dbd55 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -74,9 +74,9 @@ canal.instance.tsdb.snapshot.interval=24 # purge snapshot expire , default 360 hour(15 days) canal.instance.tsdb.snapshot.expire=360 -# rds oss binlog account -canal.instance.rds.accesskey = -canal.instance.rds.secretkey = +# aliyun ak/sk , support rds/mq +canal.aliyun.accesskey= +canal.aliyun.secretkey= ################################################# ######### destinations ############# @@ -109,5 +109,5 @@ canal.mq.bufferMemory=33554432 canal.mq.canalBatchSize=50 canal.mq.canalGetTimeout=100 canal.mq.flatMessage=true -canal.mq.compressionType= -canal.mq.acks= \ No newline at end of file +canal.mq.compressionType=none +canal.mq.acks=all \ No newline at end of file diff --git a/deployer/src/main/resources/spring/base-instance.xml b/deployer/src/main/resources/spring/base-instance.xml index e6c500b5..a89b7400 100644 --- a/deployer/src/main/resources/spring/base-instance.xml +++ b/deployer/src/main/resources/spring/base-instance.xml @@ -32,8 +32,8 @@ - - + + \ No newline at end of file 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 b9ce071b..0528d19c 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 @@ -747,7 +747,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE if (justForPositionTimestamp && logPosition.getPostion() == null && event.getWhen() > 0) { // 初始位点 entryPosition = new EntryPosition(searchBinlogFile, - event.getLogPos(), + event.getLogPos() - event.getEventLen(), event.getWhen() * 1000, event.getServerId()); logPosition.setPostion(entryPosition); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 1ba0930d..2832ff34 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -657,7 +657,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar + tableMeta.getFullName() + ", db fieldMeta : " + fieldMeta.toString() + " , binlog fieldMeta : " + info.toString() + " , on : " + event.getHeader().getLogFileName() + ":" - + event.getHeader().getLogPos()); + + (event.getHeader().getLogPos() - event.getHeader().getEventLen())); } } diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index 2e6fbe69..cf72a2dd 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -20,10 +20,10 @@ public class MQProperties { private int canalBatchSize = 50; private Long canalGetTimeout = 100L; private boolean flatMessage = true; - private String compressionType = "none"; - private String acks = "all"; - private String aliyunAccessKey = ""; - private String aliyunSecretKey = ""; + private String compressionType = "none"; + private String acks = "all"; + private String aliyunAccessKey = ""; + private String aliyunSecretKey = ""; public static class CanalDestination { From ad18e0335d738fdf856566f57ae5ff75342b9272 Mon Sep 17 00:00:00 2001 From: Alan Date: Wed, 21 Nov 2018 16:31:46 +0800 Subject: [PATCH 4/5] =?UTF-8?q?=E7=AD=89=E5=8F=B7=E4=B8=A4=E8=BE=B9?= =?UTF-8?q?=E9=83=BD=E6=9C=89=E7=A9=BA=E6=A0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 格式保持一致,看起来更像大厂出品 ^_^ --- deployer/src/main/resources/canal.properties | 54 ++++++++++---------- 1 file changed, 27 insertions(+), 27 deletions(-) diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 763dbd55..78343b43 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -1,11 +1,11 @@ ################################################# ######### common argument ############# ################################################# -canal.id= 1 -canal.ip= -canal.port=11111 -canal.metrics.pull.port=11112 -canal.zkServers= +canal.id = 1 +canal.ip = +canal.port = 11111 +canal.metrics.pull.port = 11112 +canal.zkServers = # flush data to zk canal.zookeeper.flush.period = 1000 canal.withoutNetty = false @@ -64,32 +64,32 @@ canal.instance.parser.parallel = true canal.instance.parser.parallelBufferSize = 256 # table meta tsdb info -canal.instance.tsdb.enable=true -canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:} -canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; -canal.instance.tsdb.dbUsername=canal -canal.instance.tsdb.dbPassword=canal +canal.instance.tsdb.enable = true +canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:} +canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; +canal.instance.tsdb.dbUsername = canal +canal.instance.tsdb.dbPassword = canal # dump snapshot interval, default 24 hour -canal.instance.tsdb.snapshot.interval=24 +canal.instance.tsdb.snapshot.interval = 24 # purge snapshot expire , default 360 hour(15 days) -canal.instance.tsdb.snapshot.expire=360 +canal.instance.tsdb.snapshot.expire = 360 # aliyun ak/sk , support rds/mq -canal.aliyun.accesskey= -canal.aliyun.secretkey= +canal.aliyun.accesskey = +canal.aliyun.secretkey = ################################################# ######### destinations ############# ################################################# -canal.destinations= example +canal.destinations = example # conf root dir canal.conf.dir = ../conf # auto scan instance dir add/remove and start/stop instance canal.auto.scan = true canal.auto.scan.interval = 5 -canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml -#canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml +canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml +#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml canal.instance.global.mode = spring canal.instance.global.lazy = false @@ -101,13 +101,13 @@ canal.instance.global.spring.xml = classpath:spring/file-instance.xml ################################################## ######### MQ ############# ################################################## -canal.mq.servers=127.0.0.1:6667 -canal.mq.retries=0 -canal.mq.batchSize=16384 -canal.mq.lingerMs=1 -canal.mq.bufferMemory=33554432 -canal.mq.canalBatchSize=50 -canal.mq.canalGetTimeout=100 -canal.mq.flatMessage=true -canal.mq.compressionType=none -canal.mq.acks=all \ No newline at end of file +canal.mq.servers = 127.0.0.1:6667 +canal.mq.retries = 0 +canal.mq.batchSize = 16384 +canal.mq.lingerMs = 1 +canal.mq.bufferMemory = 33554432 +canal.mq.canalBatchSize = 50 +canal.mq.canalGetTimeout = 100 +canal.mq.flatMessage = true +canal.mq.compressionType = none +canal.mq.acks = all From aa89453bf66b18102a96b95755731ad6ab13eaec Mon Sep 17 00:00:00 2001 From: agapple Date: Wed, 21 Nov 2018 22:49:22 +0800 Subject: [PATCH 5/5] fixed adapter --- .../adapter/support/CanalClientConfig.java | 56 ++++++++----------- .../launcher/config/AdapterCanalConfig.java | 2 - .../loader/AbstractCanalAdapterWorker.java | 32 +++++++---- .../launcher/loader/CanalAdapterLoader.java | 16 +++--- .../src/main/resources/application.yml | 5 +- 5 files changed, 53 insertions(+), 58 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 8e1b5596..672d83ec 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 @@ -12,25 +12,26 @@ import java.util.Map; */ public class CanalClientConfig { - private String canalServerHost; // 单机模式下canal server的 ip:port - - private String zookeeperHosts; // 集群模式下的zk地址, 如果配置了单机地址则以单机为准!! - - private String mqServers; // kafka or rocket mq 地址 - - private Boolean flatMessage = true; // 是否已flatMessage模式传输, 只适用于mq模式 - - private Integer batchSize; // 批大小 - - private Integer syncBatchSize = 1000; // 同步分批提交大小 - - private Integer retries; // 重试次数 - - private Long timeout; // 消费超时时间 - - private String mode = "tcp"; // 模式 tcp kafka rocketMQ - - private List canalAdapters; // canal adapters 配置 + // 单机模式下canal server的ip:port + private String canalServerHost; + // 集群模式下的zk地址,如果配置了单机地址则以单机为准!! + private String zookeeperHosts; + // kafka or rocket mq 地址 + private String mqServers; + // 是否已flatMessage模式传输,只适用于mq模式 + private Boolean flatMessage = true; + // 批大小 + private Integer batchSize; + // 同步分批提交大小 + private Integer syncBatchSize = 1000; + // 重试次数 + private Integer retries; + // 消费超时时间 + private Long timeout; + // 模式 tcp kafka rocketMQ + private String mode = "tcp"; + // canal adapters 配置 + private List canalAdapters; public String getCanalServerHost() { return canalServerHost; @@ -116,9 +117,7 @@ public class CanalClientConfig { private String instance; // 实例名 - private String topic; // mq topic - - private List groups; // 适配器分组列表 + private List groups; // 适配器分组列表 public String getInstance() { return instance; @@ -137,22 +136,13 @@ public class CanalClientConfig { public void setGroups(List groups) { this.groups = groups; } - - public String getTopic() { - return topic; - } - - public void setTopic(String topic) { - this.topic = topic; - } } public static class Group { - private String groupId; // group id - + // group id + private String groupId = "default"; private List outerAdapters; // 适配器列表 - private Map outerAdaptersMap = new LinkedHashMap<>(); public String getGroupId() { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java index de27296e..2b9bb243 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java @@ -37,8 +37,6 @@ public class AdapterCanalConfig extends CanalClientConfig { for (CanalAdapter canalAdapter : canalAdapters) { if (canalAdapter.getInstance() != null) { DESTINATIONS.add(canalAdapter.getInstance()); - } else if (canalAdapter.getTopic() != null) { - DESTINATIONS.add(canalAdapter.getInstance()); } } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java index 3702fd1e..db87bf01 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AbstractCanalAdapterWorker.java @@ -2,9 +2,12 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.*; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -12,6 +15,7 @@ import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; import com.alibaba.otter.canal.client.CanalMQConnector; import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.MessageUtil; import com.alibaba.otter.canal.protocol.FlatMessage; @@ -180,18 +184,22 @@ public abstract class AbstractCanalAdapterWorker { */ private void batchSync(List dmls, OuterAdapter adapter) { // 分批同步 - int len = 0; - List dmlsBatch = new ArrayList<>(); - for (Dml dml : dmls) { - dmlsBatch.add(dml); - len += dml.getData().size(); - if (len >= canalClientConfig.getSyncBatchSize()) { - adapter.sync(dmlsBatch); - dmlsBatch.clear(); - len = 0; + if (dmls.size() <= canalClientConfig.getSyncBatchSize()) { + adapter.sync(dmls); + } else { + int len = 0; + List dmlsBatch = new ArrayList<>(); + for (Dml dml : dmls) { + dmlsBatch.add(dml); + len += dml.getData().size(); + if (len >= canalClientConfig.getSyncBatchSize()) { + adapter.sync(dmlsBatch); + dmlsBatch.clear(); + len = 0; + } } + adapter.sync(dmlsBatch); } - adapter.sync(dmlsBatch); } public void start() { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index a72197fd..162cdc11 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -97,14 +97,14 @@ public class CanalAdapterLoader { CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig, canalClientConfig.getMqServers(), - canalAdapter.getTopic(), + canalAdapter.getInstance(), group.getGroupId(), canalOuterAdapterGroups, canalClientConfig.getFlatMessage()); - canalMQWorker.put(canalAdapter.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker); + canalMQWorker.put(canalAdapter.getInstance() + "-kafka-" + group.getGroupId(), canalKafkaWorker); canalKafkaWorker.start(); - logger.info("Start adapter for canal-client mq topic: {} succeed", - canalAdapter.getTopic() + "-" + group.getGroupId()); + logger.info("Start adapter for canal-client mq topic: {} succeed", canalAdapter.getInstance() + "-" + + group.getGroupId()); } } } else if ("rocketMQ".equalsIgnoreCase(canalClientConfig.getMode())) { @@ -119,15 +119,15 @@ public class CanalAdapterLoader { canalOuterAdapterGroups.add(canalOuterAdapters); CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig, canalClientConfig.getMqServers(), - canalAdapter.getTopic(), + canalAdapter.getInstance(), group.getGroupId(), canalOuterAdapterGroups, canalClientConfig.getFlatMessage()); - canalMQWorker.put(canalAdapter.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); + canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); rocketMQWorker.start(); - logger.info("Start adapter for canal-client mq topic: {} succeed", - canalAdapter.getTopic() + "-" + group.getGroupId()); + logger.info("Start adapter for canal-client mq topic: {} succeed", canalAdapter.getInstance() + "-" + + group.getGroupId()); } } } diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 4d120c6a..5495a4f8 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -28,10 +28,9 @@ canal.conf: # username: root # password: 121212 canalAdapters: - - instance: example -# topic: example + - instance: example # canal instance Name or mq topic name groups: - - #groupId: g1 + - groupId: g1 outerAdapters: - name: logger # - name: rdb