From 65b5325ca63e8a033a92823e7eacd7e193f1e2fe Mon Sep 17 00:00:00 2001 From: shicongyang Date: Tue, 15 Oct 2019 21:32:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BB=BA=E8=AE=AE=E5=A2=9E=E5=8A=A0=E4=B8=80?= =?UTF-8?q?=E4=B8=AA=E5=B1=9E=E6=80=A7=EF=BC=8C=E9=92=88=E5=AF=B9=E6=A0=B9?= =?UTF-8?q?=E6=8D=AE=E6=9F=90=E4=B8=80=E5=AD=97=E6=AE=B5hash=E6=97=B6?= =?UTF-8?q?=EF=BC=8Chash=E7=AE=97=E6=B3=95=E5=8F=AF=E4=BB=A5=E4=B8=8D?= =?UTF-8?q?=E6=A0=B9=E6=8D=AEdatabase=E7=AE=97=E5=87=BA=20#2248=20(#2294)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/deployer/CanalConstants.java | 1 + .../otter/canal/deployer/CanalStarter.java | 5 +++++ deployer/src/main/resources/canal.properties | 1 + .../otter/canal/common/MQMessageUtils.java | 16 ++++++++++++---- .../alibaba/otter/canal/common/MQProperties.java | 14 +++++++++++++- .../otter/canal/kafka/CanalKafkaProducer.java | 9 +++++++-- .../canal/rocketmq/CanalRocketMQProducer.java | 7 +++++-- 7 files changed, 44 insertions(+), 9 deletions(-) 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 77b6405d..4769c20f 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 @@ -75,6 +75,7 @@ public class CanalConstants { public static final String CANAL_MQ_VHOST = ROOT + "." + "mq.vhost"; public static final String CANAL_MQ_ALIYUN_UID = ROOT + "." + "mq.aliyunuid"; public static final String CANAL_MQ_EXCHANGE = ROOT + "." + "mq.exchange"; + public static final String CANAL_MQ_DATABASE_HASH = ROOT + "." + "mq.database.hash"; 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/CanalStarter.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java index ccc3a82c..64fd191a 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java @@ -298,6 +298,11 @@ public class CanalStarter { mqProperties.setExchange(exchange); } + String databaseHash = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_DATABASE_HASH); + if (!StringUtils.isEmpty(databaseHash)){ + mqProperties.setDatabaseHash(Boolean.valueOf(databaseHash)); + } + for (Object key : properties.keySet()) { key = StringUtils.trim(key.toString()); if (((String) key).startsWith(CanalConstants.CANAL_MQ_PROPERTIES)) { diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 58ba13c5..9b464324 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -137,6 +137,7 @@ canal.mq.username= canal.mq.password= canal.mq.aliyunuid= +canal.mq.database.hash = true ################################################## ######### Kafka Kerberos Info ############# ################################################## diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index ae966260..c38d03dd 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -234,10 +234,11 @@ public class MQMessageUtils { * * @param partitionsNum 分区数 * @param pkHashConfigs 分区库表主键正则表达式 + * @param databaseHash 是否取消根据database进行hash * @return 分区message数组 */ @SuppressWarnings("unchecked") - public static Message[] messagePartition(EntryRowData[] datas, long id, Integer partitionsNum, String pkHashConfigs) { + public static Message[] messagePartition(EntryRowData[] datas, long id, Integer partitionsNum, String pkHashConfigs,boolean databaseHash) { if (partitionsNum == null) { partitionsNum = 1; } @@ -279,7 +280,10 @@ public class MQMessageUtils { RowChange.Builder rowChangeBuilder = RowChange.newBuilder(rowChange); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - int hashCode = database.hashCode(); + int hashCode = 0; + if(databaseHash){ + hashCode = database.hashCode(); + } CanalEntry.EventType eventType = rowChange.getEventType(); List columns = null; if (eventType == CanalEntry.EventType.DELETE) { @@ -444,9 +448,10 @@ public class MQMessageUtils { * @param flatMessage flatMessage * @param partitionsNum 分区数量 * @param pkHashConfigs hash映射 + * @param databaseHash 是否取消根据database进行hash * @return 拆分后的flatMessage数组 */ - public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs) { + public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs,boolean databaseHash) { if (partitionsNum == null) { partitionsNum = 1; } @@ -476,7 +481,10 @@ public class MQMessageUtils { int idx = 0; for (Map row : flatMessage.getData()) { - int hashCode = database.hashCode(); + int hashCode = 0; + if(databaseHash){ + hashCode = database.hashCode(); + } if (pkNames != null) { for (String pkName : pkNames) { String value = row.get(pkName); 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 df9a3bfe..7dcea47f 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 @@ -48,6 +48,9 @@ public class MQProperties { private String exchange = ""; // 消息发送的并行度 private int parallelThreadSize = 8; + //是否取消根据database进行hash + private boolean databaseHash = true; + public static class CanalDestination { @@ -339,6 +342,13 @@ public class MQProperties { this.parallelThreadSize = parallelThreadSize; } + public boolean getDatabaseHash() { + return databaseHash; + } + + public void setDatabaseHash(boolean databaseHash) { + this.databaseHash = databaseHash; + } @Override public String toString() { return "MQProperties [servers=" + servers + ", retries=" + retries + ", batchSize=" + batchSize + ", lingerMs=" @@ -352,7 +362,9 @@ public class MQProperties { + kerberosEnable + ", kerberosKrb5FilePath=" + kerberosKrb5FilePath + ", kerberosJaasFilePath=" + kerberosJaasFilePath + ", username=" + username + ", password=" + password + ", vhost=" + vhost + ", aliyunUID=" + aliyunUID + ", exchange=" + exchange + ", parallelThreadSize=" + parallelThreadSize - + "]"; + + ",databaseHash=" + databaseHash +"]"; } + + } 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 be9fa53b..ef1148da 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 @@ -42,11 +42,14 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro private Producer producer; private MQProperties kafkaProperties; + private boolean databaseHash; + @Override public void init(MQProperties kafkaProperties) { super.init(kafkaProperties); this.kafkaProperties = kafkaProperties; + databaseHash = kafkaProperties.getDatabaseHash(); Properties properties = new Properties(); properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("acks", kafkaProperties.getAcks()); @@ -172,7 +175,8 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro Message[] messages = MQMessageUtils.messagePartition(datas, message.getId(), canalDestination.getPartitionsNum(), - canalDestination.getPartitionHash()); + canalDestination.getPartitionHash(), + databaseHash); int length = messages.length; for (int i = 0; i < length; i++) { Message messagePartition = messages[i]; @@ -201,7 +205,8 @@ public class CanalKafkaProducer extends AbstractMQProducer implements CanalMQPro if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, canalDestination.getPartitionsNum(), - canalDestination.getPartitionHash()); + canalDestination.getPartitionHash(), + databaseHash); int length = partitionFlatMessage.length; for (int i = 0; i < length; i++) { FlatMessage flatMessagePart = partitionFlatMessage[i]; 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 50c6a6fd..0d609695 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 @@ -40,10 +40,13 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ private MQProperties mqProperties; private static final String CLOUD_ACCESS_CHANNEL = "cloud"; + private boolean databaseHash; + @Override public void init(MQProperties rocketMQProperties) { super.init(rocketMQProperties); this.mqProperties = rocketMQProperties; + this.databaseHash = rocketMQProperties.getDatabaseHash(); RPCHook rpcHook = null; if (rocketMQProperties.getAliyunAccessKey().length() > 0 && rocketMQProperties.getAliyunSecretKey().length() > 0) { @@ -125,7 +128,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas, message.getId(), destination.getPartitionsNum(), - destination.getPartitionHash()); + destination.getPartitionHash(),databaseHash); int length = messages.length; ExecutorTemplate template = new ExecutorTemplate(executor); @@ -168,7 +171,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ for (FlatMessage flatMessage : flatMessages) { FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, destination.getPartitionsNum(), - destination.getPartitionHash()); + destination.getPartitionHash(),databaseHash); int length = partitionFlatMessage.length; for (int i = 0; i < length; i++) { partitionFlatMessages.get(i).add(partitionFlatMessage[i]);