From b81e9dfeb860ac03e9c693ef01269bcbc169ca56 Mon Sep 17 00:00:00 2001 From: liguangyao <1012461549@qq.com> Date: Sat, 22 Aug 2020 13:11:53 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E4=B8=8D=E5=90=8Ctopic?= =?UTF-8?q?=E6=8C=87=E5=AE=9A=E4=B8=93=E5=B1=9EpartitionNum=20(#2479)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/producer/MQDestination.java | 9 +++ .../core/producer/MQMessageUtils.java | 57 +++++++++++++++++++ .../kafka/producer/CanalKafkaProducer.java | 9 ++- .../producer/CanalRocketMQProducer.java | 9 ++- .../resources/example/instance.properties | 1 + .../resources/spring/default-instance.xml | 1 + .../main/resources/spring/file-instance.xml | 1 + .../main/resources/spring/group-instance.xml | 1 + .../main/resources/spring/memory-instance.xml | 1 + .../canal/instance/core/CanalMQConfig.java | 9 +++ .../otter/canal/server/CanalMQStarter.java | 1 + 11 files changed, 95 insertions(+), 4 deletions(-) diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQDestination.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQDestination.java index 00d8b870..6b45d04f 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQDestination.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQDestination.java @@ -14,6 +14,7 @@ public class MQDestination { private Integer partitionsNum; private String partitionHash; private String dynamicTopic; + private String dynamicTopicPartitionNum; public String getCanalDestination() { return canalDestination; @@ -62,4 +63,12 @@ public class MQDestination { public void setDynamicTopic(String dynamicTopic) { this.dynamicTopic = dynamicTopic; } + + public String getDynamicTopicPartitionNum() { + return dynamicTopicPartitionNum; + } + + public void setDynamicTopicPartitionNum(String dynamicTopicPartitionNum) { + this.dynamicTopicPartitionNum = dynamicTopicPartitionNum; + } } diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java index ad2b4c87..1c3b39c8 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQMessageUtils.java @@ -103,6 +103,38 @@ public class MQMessageUtils { } }); + private static Map> topicPartitionDatas = MigrateMap.makeComputingMap(CacheBuilder.newBuilder() + .softValues(), + new Function>() { + + public List apply(String tPConfigs) { + List datas = Lists.newArrayList(); + String[] tPArray = StringUtils.split(StringUtils.replace(tPConfigs, + ",", + ";"), + ";"); + for (String tPConfig : tPArray) { + TopicPartitionData data = new TopicPartitionData(); + int i = tPConfig.lastIndexOf(":"); + if (i > 0) { + String tStr = tPConfig.substring(0, i); + String pStr = tPConfig.substring(i + 1); + if (!isWildCard(tStr)) { + data.simpleName = tStr; + } else { + data.regexFilter = new AviaterRegexFilter(tStr); + } + if (!StringUtils.isEmpty(pStr) && StringUtils.isNumeric(pStr)) { + data.partitionNum = Integer.valueOf(pStr); + } + datas.add(data); + } + } + + return datas; + } + }); + /** * 按 schema 或者 schema+table 将 message 分配到对应topic * @@ -619,6 +651,24 @@ public class MQMessageUtils { return false; } + public static Integer parseDynamicTopicPartition(String name, String tPConfigs) { + if (!StringUtils.isEmpty(tPConfigs)) { + List datas = topicPartitionDatas.get(tPConfigs); + for (TopicPartitionData data : datas) { + if (data.simpleName != null) { + if (data.simpleName.equalsIgnoreCase(name)) { + return data.partitionNum; + } + } else { + if (data.regexFilter.filter(name)) { + return data.partitionNum; + } + } + } + } + return null; + } + private static boolean isWildCard(String value) { // not contaiins '.' ? return StringUtils.containsAny(value, new char[] { '*', '?', '+', '|', '(', ')', '{', '}', '[', ']', '\\', '$', @@ -656,6 +706,13 @@ public class MQMessageUtils { public AviaterRegexFilter tableRegexFilter; } + public static class TopicPartitionData { + + public String simpleName; + public AviaterRegexFilter regexFilter; + public Integer partitionNum; + } + public static class EntryRowData { public Entry entry; diff --git a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java index df50bc28..6881da38 100644 --- a/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java +++ b/connector/kafka-connector/src/main/java/com/alibaba/otter/canal/connector/kafka/producer/CanalKafkaProducer.java @@ -196,6 +196,11 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro private List send(MQDestination mqDestination, String topicName, Message message, boolean flat) { List> records = new ArrayList<>(); + // 获取当前topic的分区数 + Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName, mqDestination.getDynamicTopicPartitionNum()); + if (partitionNum == null) { + partitionNum = mqDestination.getPartitionsNum(); + } if (!flat) { if (mqDestination.getPartitionHash() != null && !mqDestination.getPartitionHash().isEmpty()) { // 并发构造 @@ -203,7 +208,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro // 串行分区 Message[] messages = MQMessageUtils.messagePartition(datas, message.getId(), - mqDestination.getPartitionsNum(), + partitionNum, mqDestination.getPartitionHash(), this.mqProperties.isDatabaseHash()); int length = messages.length; @@ -233,7 +238,7 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro for (FlatMessage flatMessage : flatMessages) { if (mqDestination.getPartitionHash() != null && !mqDestination.getPartitionHash().isEmpty()) { FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, - mqDestination.getPartitionsNum(), + partitionNum, mqDestination.getPartitionHash(), this.mqProperties.isDatabaseHash()); int length = partitionFlatMessage.length; diff --git a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java index fd2fd304..bbd42316 100644 --- a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java +++ b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java @@ -161,6 +161,11 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ } public void send(final MQDestination destination, String topicName, com.alibaba.otter.canal.protocol.Message message) { + // 获取当前topic的分区数 + Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName, destination.getDynamicTopicPartitionNum()); + if (partitionNum == null) { + partitionNum = destination.getPartitionsNum(); + } if (!mqProperties.isFlatMessage()) { if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { // 并发构造 @@ -168,7 +173,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ // 串行分区 com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas, message.getId(), - destination.getPartitionsNum(), + partitionNum, destination.getPartitionHash(), mqProperties.isDatabaseHash()); int length = messages.length; @@ -207,7 +212,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ for (FlatMessage flatMessage : flatMessages) { FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, - destination.getPartitionsNum(), + partitionNum, destination.getPartitionHash(), mqProperties.isDatabaseHash()); int length = partitionFlatMessage.length; diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 06e3dad3..c41affe0 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -54,4 +54,5 @@ canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=3 #canal.mq.partitionHash=test.table:id^name,.*\\..* +#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6 ################################################# diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index effb9cac..95f02bd0 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -200,5 +200,6 @@ + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index 05c1d04d..b36d87d4 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -186,5 +186,6 @@ + diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index a792d90d..8d297331 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -276,5 +276,6 @@ + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index 6d65ad44..19ec00c9 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -174,5 +174,6 @@ + diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java index 3d49f0fa..fbf57ea1 100644 --- a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java @@ -7,6 +7,7 @@ public class CanalMQConfig { private Integer partitionsNum; private String partitionHash; private String dynamicTopic; + private String dynamicTopicPartitionNum; public String getTopic() { return topic; @@ -47,4 +48,12 @@ public class CanalMQConfig { public void setDynamicTopic(String dynamicTopic) { this.dynamicTopic = dynamicTopic; } + + public String getDynamicTopicPartitionNum() { + return dynamicTopicPartitionNum; + } + + public void setDynamicTopicPartitionNum(String dynamicTopicPartitionNum) { + this.dynamicTopicPartitionNum = dynamicTopicPartitionNum; + } } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index 50be0d45..d368f4a3 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -160,6 +160,7 @@ public class CanalMQStarter { canalDestination.setDynamicTopic(mqConfig.getDynamicTopic()); canalDestination.setPartitionsNum(mqConfig.getPartitionsNum()); canalDestination.setPartitionHash(mqConfig.getPartitionHash()); + canalDestination.setDynamicTopicPartitionNum(mqConfig.getDynamicTopicPartitionNum()); canalServer.subscribe(clientIdentity); logger.info("## the MQ producer: {} is running now ......", destination);