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 6b45d04f..bdf08587 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
@@ -15,6 +15,7 @@ public class MQDestination {
private String partitionHash;
private String dynamicTopic;
private String dynamicTopicPartitionNum;
+ private Boolean enableDynamicQueuePartition;
public String getCanalDestination() {
return canalDestination;
@@ -71,4 +72,12 @@ public class MQDestination {
public void setDynamicTopicPartitionNum(String dynamicTopicPartitionNum) {
this.dynamicTopicPartitionNum = dynamicTopicPartitionNum;
}
+
+ public Boolean getEnableDynamicQueuePartition() {
+ return enableDynamicQueuePartition;
+ }
+
+ public void setEnableDynamicQueuePartition(Boolean enableDynamicQueuePartition) {
+ this.enableDynamicQueuePartition = enableDynamicQueuePartition;
+ }
}
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 74d80671..b32d2bde 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
@@ -1,29 +1,5 @@
package com.alibaba.otter.canal.connector.rocketmq.producer;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.rocketmq.acl.common.AclClientRPCHook;
-import org.apache.rocketmq.acl.common.SessionCredentials;
-import org.apache.rocketmq.client.AccessChannel;
-import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
-import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
-import org.apache.rocketmq.client.producer.SendResult;
-import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.remoting.RPCHook;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.otter.canal.common.CanalException;
@@ -40,6 +16,29 @@ import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQConstants;
import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQProducerConfig;
import com.alibaba.otter.canal.protocol.FlatMessage;
+import org.apache.commons.lang.StringUtils;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.client.AccessChannel;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
/**
* RocketMQ Producer SPI 实现
@@ -186,6 +185,12 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
// 获取当前topic的分区数
Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName,
destination.getDynamicTopicPartitionNum());
+
+ // 获取topic的队列数为分区数
+ if(partitionNum == null){
+ partitionNum = getTopicDynamicQueuesSize(destination.getEnableDynamicQueuePartition(), topicName);
+ }
+
if (partitionNum == null) {
partitionNum = destination.getPartitionsNum();
}
@@ -353,4 +358,18 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
super.stop();
}
+
+ private Integer getTopicDynamicQueuesSize(Boolean enable, String topicName){
+ if(enable!=null && enable){
+ topicName = this.defaultMQProducer.withNamespace(topicName);
+ DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl();
+ TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topicName);
+ if(topicInfo == null){
+ return null;
+ }else{
+ return topicInfo.getMessageQueueList().size();
+ }
+ }
+ return null;
+ }
}
diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml
index add43b74..9907ae80 100644
--- a/deployer/src/main/resources/spring/default-instance.xml
+++ b/deployer/src/main/resources/spring/default-instance.xml
@@ -206,5 +206,6 @@
+
diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml
index 65fab920..800f9812 100644
--- a/deployer/src/main/resources/spring/file-instance.xml
+++ b/deployer/src/main/resources/spring/file-instance.xml
@@ -192,5 +192,6 @@
+
diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml
index bd808b24..b0b887f9 100644
--- a/deployer/src/main/resources/spring/group-instance.xml
+++ b/deployer/src/main/resources/spring/group-instance.xml
@@ -287,5 +287,6 @@
+
diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml
index 5ec293e0..a7dc634c 100644
--- a/deployer/src/main/resources/spring/memory-instance.xml
+++ b/deployer/src/main/resources/spring/memory-instance.xml
@@ -180,5 +180,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 fbf57ea1..af362271 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
@@ -8,6 +8,7 @@ public class CanalMQConfig {
private String partitionHash;
private String dynamicTopic;
private String dynamicTopicPartitionNum;
+ private Boolean enableDynamicQueuePartition;
public String getTopic() {
return topic;
@@ -56,4 +57,12 @@ public class CanalMQConfig {
public void setDynamicTopicPartitionNum(String dynamicTopicPartitionNum) {
this.dynamicTopicPartitionNum = dynamicTopicPartitionNum;
}
+
+ public Boolean getEnableDynamicQueuePartition() {
+ return enableDynamicQueuePartition;
+ }
+
+ public void setEnableDynamicQueuePartition(Boolean enableDynamicQueuePartition) {
+ this.enableDynamicQueuePartition = enableDynamicQueuePartition;
+ }
}
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 3fd9f861..58f7cd66 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
@@ -1,17 +1,5 @@
package com.alibaba.otter.canal.server;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.slf4j.MDC;
-
import com.alibaba.otter.canal.connector.core.config.MQProperties;
import com.alibaba.otter.canal.connector.core.producer.MQDestination;
import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
@@ -21,6 +9,17 @@ import com.alibaba.otter.canal.instance.core.CanalMQConfig;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
public class CanalMQStarter {
@@ -157,6 +156,7 @@ public class CanalMQStarter {
canalDestination.setPartitionsNum(mqConfig.getPartitionsNum());
canalDestination.setPartitionHash(mqConfig.getPartitionHash());
canalDestination.setDynamicTopicPartitionNum(mqConfig.getDynamicTopicPartitionNum());
+ canalDestination.setEnableDynamicQueuePartition(mqConfig.getEnableDynamicQueuePartition());
canalServer.subscribe(clientIdentity);
logger.info("## the MQ producer: {} is running now ......", destination);