From 03ee9c7ecf8120b5fe8d47ed2c0b3b64dc376e01 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 00:09:49 +0800 Subject: [PATCH] =?UTF-8?q?=E6=A0=B9=E6=8D=AE=E4=B8=BB=E9=94=AEHash?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=AD=A3=E5=88=99=E8=A1=A8=E8=BE=BE=E5=BC=8F?= =?UTF-8?q?=E6=94=AF=E6=8C=81=EF=BC=8Cpk=E9=80=9A=E9=85=8D=E7=AC=A6?= =?UTF-8?q?=E6=8C=87=E5=AE=9A=E8=87=AA=E5=8A=A8=E8=8E=B7=E5=8F=96=E6=94=AF?= =?UTF-8?q?=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/instance/core/CanalMQConfig.java | 34 +--- .../otter/canal/protocol/FlatMessage.java | 168 ++++++++++++------ .../otter/canal/common/MQProperties.java | 15 +- .../otter/canal/server/CanalMQStarter.java | 2 +- 4 files changed, 123 insertions(+), 96 deletions(-) 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 c275f150..53c50742 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 @@ -1,16 +1,11 @@ package com.alibaba.otter.canal.instance.core; -import java.util.LinkedHashMap; -import java.util.Map; - public class CanalMQConfig { - private String topic; - private Integer partition; - private Integer partitionsNum; - private String partitionHash; - - private volatile Map partitionHashProperties; + private String topic; + private Integer partition; + private Integer partitionsNum; + private String partitionHash; public String getTopic() { return topic; @@ -44,25 +39,4 @@ public class CanalMQConfig { this.partitionHash = partitionHash; } - public Map getPartitionHashProperties() { - if (partitionHashProperties == null) { - synchronized (CanalMQConfig.class) { - if (partitionHashProperties == null) { - if (partitionHash != null) { - partitionHashProperties = new LinkedHashMap<>(); - String[] items = partitionHash.split(","); - for (String item : items) { - int i = item.indexOf(":"); - if (i > -1) { - String dbTable = item.substring(0, i).trim(); - String pk = item.substring(i + 1).trim(); - partitionHashProperties.put(dbTable, pk); - } - } - } - } - } - } - return partitionHashProperties; - } } 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 94c539cb..568d689a 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,12 +1,10 @@ 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 java.util.regex.Pattern; + +import org.apache.commons.lang.StringUtils; import com.google.protobuf.ByteString; @@ -33,8 +31,7 @@ public class FlatMessage implements Serializable { private List> data; private List> old; - public FlatMessage(){ - } + private transient Message message; // 所属message public FlatMessage(long id){ this.id = id; @@ -136,6 +133,14 @@ public class FlatMessage implements Serializable { this.es = es; } + public Message getMessage() { + return message; + } + + public void setMessage(Message message) { + this.message = message; + } + /** * 将Message转换为FlatMessage * @@ -187,6 +192,7 @@ public class FlatMessage implements Serializable { flatMessage.setEs(entry.getHeader().getExecuteTime()); flatMessage.setTs(System.currentTimeMillis()); flatMessage.setSql(rowChange.getSql()); + flatMessage.setMessage(message); if (!rowChange.getIsDdl()) { Map sqlType = new LinkedHashMap<>(); @@ -269,70 +275,116 @@ public class FlatMessage implements Serializable { * * @param flatMessage flatMessage * @param partitionsNum 分区数量 - * @param pkHashConfig hash映射 + * @param pkHashConfigs hash映射 * @return 拆分后的flatMessage数组 */ - public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, - Map pkHashConfig) { + public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs) { if (partitionsNum == null) { partitionsNum = 1; } FlatMessage[] partitionMessages = new FlatMessage[partitionsNum]; - String pk = pkHashConfig.get(flatMessage.getDatabase() + "." + flatMessage.getTable()); - if (pk == null || flatMessage.getIsDdl()) { + + if (flatMessage.getIsDdl()) { partitionMessages[0] = flatMessage; } else { if (flatMessage.getData() != null) { - int idx = 0; - for (Map row : flatMessage.getData()) { - Map o = null; - if (flatMessage.getOld() != null) { - o = flatMessage.getOld().get(idx); - } - String value; - // 如果old中有pk值说明主键有修改, 以旧的主键值hash为准 - if (o != null && o.containsKey(pk)) { - value = o.get(pk); - } else { - value = row.get(pk); - } - if (value == null) { - value = ""; - } - int hash = value.hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序 - pkHash = Math.abs(pkHash); + String database = flatMessage.getDatabase(); + String table = flatMessage.getTable(); - FlatMessage flatMessageTmp = partitionMessages[pkHash]; - if (flatMessageTmp == null) { - flatMessageTmp = new FlatMessage(flatMessage.getId()); - partitionMessages[pkHash] = flatMessageTmp; - flatMessageTmp.setDatabase(flatMessage.getDatabase()); - flatMessageTmp.setTable(flatMessage.getTable()); - flatMessageTmp.setIsDdl(flatMessage.getIsDdl()); - flatMessageTmp.setType(flatMessage.getType()); - flatMessageTmp.setSql(flatMessage.getSql()); - flatMessageTmp.setSqlType(flatMessage.getSqlType()); - flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); - flatMessageTmp.setEs(flatMessage.getEs()); - flatMessageTmp.setTs(flatMessage.getTs()); + String pk = null; + boolean isMatch = false; + + String[] pkHashConfigArray = StringUtils.split(pkHashConfigs, ","); + for (String pkHashConfig : pkHashConfigArray) { + int i = pkHashConfig.lastIndexOf("."); + if (!pkHashConfig.endsWith(".$pk$")) { + // 如果指定了主键 + pk = pkHashConfig.substring(i + 1); } - List> data = flatMessageTmp.getData(); - if (data == null) { - data = new ArrayList<>(); - flatMessageTmp.setData(data); + pkHashConfig = pkHashConfig.substring(0, i); + isMatch = Pattern.matches(pkHashConfig, database + "." + table); + if (isMatch) { + break; } - data.add(row); - if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) { - List> old = flatMessageTmp.getOld(); - if (old == null) { - old = new ArrayList<>(); - flatMessageTmp.setOld(old); + } + + if (!isMatch) { + // 如果都没有匹配，发送到第一个分区 + partitionMessages[0] = flatMessage; + } else { + if (pk == null) { + // 如果未指定主键(通配符主键)，从原生message中取主键字段 + CanalEntry.Entry entry = flatMessage.getMessage().getEntries().get(0); + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + pk = column.getName(); + break; + } + } + } + if (pk == null || !flatMessage.getData().get(0).containsKey(pk)) { + // 如果都没有匹配的主键，发送到第一个分区 + partitionMessages[0] = flatMessage; + } else { + int idx = 0; + for (Map row : flatMessage.getData()) { + Map o = null; + if (flatMessage.getOld() != null) { + o = flatMessage.getOld().get(idx); + } + String value; + // 如果old中有pk值说明主键有修改, 以旧的主键值hash为准 + if (o != null && o.containsKey(pk)) { + value = o.get(pk); + } else { + value = row.get(pk); + } + if (value == null) { + value = ""; + } + int hash = value.hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序 + pkHash = Math.abs(pkHash); + + FlatMessage flatMessageTmp = partitionMessages[pkHash]; + if (flatMessageTmp == null) { + flatMessageTmp = new FlatMessage(flatMessage.getId()); + partitionMessages[pkHash] = flatMessageTmp; + flatMessageTmp.setDatabase(flatMessage.getDatabase()); + flatMessageTmp.setTable(flatMessage.getTable()); + flatMessageTmp.setIsDdl(flatMessage.getIsDdl()); + flatMessageTmp.setType(flatMessage.getType()); + flatMessageTmp.setSql(flatMessage.getSql()); + flatMessageTmp.setSqlType(flatMessage.getSqlType()); + flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); + flatMessageTmp.setEs(flatMessage.getEs()); + flatMessageTmp.setTs(flatMessage.getTs()); + } + List> data = flatMessageTmp.getData(); + if (data == null) { + data = new ArrayList<>(); + flatMessageTmp.setData(data); + } + data.add(row); + if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) { + List> old = flatMessageTmp.getOld(); + if (old == null) { + old = new ArrayList<>(); + flatMessageTmp.setOld(old); + } + old.add(flatMessage.getOld().get(idx)); + } + idx++; } - old.add(flatMessage.getOld().get(idx)); } - idx++; } } } 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 a337ccc6..7e82f9c6 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 @@ -28,11 +28,11 @@ public class MQProperties { public static class CanalDestination { - private String canalDestination; - private String topic; - private Integer partition; - private Integer partitionsNum; - private Map partitionHash; + private String canalDestination; + private String topic; + private Integer partition; + private Integer partitionsNum; + private String partitionHash; public String getCanalDestination() { return canalDestination; @@ -66,11 +66,11 @@ public class MQProperties { this.partitionsNum = partitionsNum; } - public Map getPartitionHash() { + public String getPartitionHash() { return partitionHash; } - public void setPartitionHash(Map partitionHash) { + public void setPartitionHash(String partitionHash) { this.partitionHash = partitionHash; } } @@ -186,6 +186,7 @@ public class MQProperties { public void setAliyunSecretKey(String aliyunSecretKey) { this.aliyunSecretKey = aliyunSecretKey; } + public int getMaxRequestSize() { return maxRequestSize; } 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 a283e23f..248e5af4 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 @@ -138,7 +138,7 @@ public class CanalMQStarter { canalDestination.setTopic(mqConfig.getTopic()); canalDestination.setPartition(mqConfig.getPartition()); canalDestination.setPartitionsNum(mqConfig.getPartitionsNum()); - canalDestination.setPartitionHash(mqConfig.getPartitionHashProperties()); + canalDestination.setPartitionHash(mqConfig.getPartitionHash()); canalServer.subscribe(clientIdentity); logger.info("## the MQ producer: {} is running now ......", destination);