From 03ee9c7ecf8120b5fe8d47ed2c0b3b64dc376e01 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 00:09:49 +0800 Subject: [PATCH 1/7] =?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); From 547cbf82c83eb4987e9568569788818fe1e7a806 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 01:03:18 +0800 Subject: [PATCH 2/7] =?UTF-8?q?=E5=AF=B9=E5=8E=9F=E7=94=9Fmessage=E5=8F=91?= =?UTF-8?q?=E9=80=81mq=E7=9A=84=E5=88=86=E5=8C=BA=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/protocol/FlatMessage.java | 42 +++--- .../alibaba/otter/canal/protocol/Message.java | 113 +++++++++++++++++ .../otter/canal/kafka/CanalKafkaProducer.java | 30 +++-- .../canal/rocketmq/CanalRocketMQProducer.java | 120 ++++++++++++------ 4 files changed, 240 insertions(+), 65 deletions(-) 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 568d689a..05d8f795 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 @@ -14,24 +14,24 @@ import com.google.protobuf.ByteString; */ public class FlatMessage implements Serializable { - private static final long serialVersionUID = -3386650678735860050L; + private static final long serialVersionUID = -3386650678735860050L; - private long id; - private String database; - private String table; - private Boolean isDdl; - private String type; + private long id; + private String database; + private String table; + private Boolean isDdl; + private String type; // binlog executeTime - private Long es; + private Long es; // dml build timeStamp - private Long ts; - private String sql; - private Map sqlType; - private Map mysqlType; - private List> data; - private List> old; + private Long ts; + private String sql; + private Map sqlType; + private Map mysqlType; + private List> data; + private List> old; - private transient Message message; // 所属message + private transient CanalEntry.Entry entry; // 所属entry public FlatMessage(long id){ this.id = id; @@ -133,12 +133,12 @@ public class FlatMessage implements Serializable { this.es = es; } - public Message getMessage() { - return message; + public CanalEntry.Entry getEntry() { + return entry; } - public void setMessage(Message message) { - this.message = message; + public void setEntry(CanalEntry.Entry entry) { + this.entry = entry; } /** @@ -192,7 +192,7 @@ public class FlatMessage implements Serializable { flatMessage.setEs(entry.getHeader().getExecuteTime()); flatMessage.setTs(System.currentTimeMillis()); flatMessage.setSql(rowChange.getSql()); - flatMessage.setMessage(message); + flatMessage.setEntry(entry); if (!rowChange.getIsDdl()) { Map sqlType = new LinkedHashMap<>(); @@ -287,7 +287,7 @@ public class FlatMessage implements Serializable { if (flatMessage.getIsDdl()) { partitionMessages[0] = flatMessage; } else { - if (flatMessage.getData() != null) { + if (flatMessage.getData() != null && !flatMessage.getData().isEmpty()) { String database = flatMessage.getDatabase(); String table = flatMessage.getTable(); @@ -314,7 +314,7 @@ public class FlatMessage implements Serializable { } else { if (pk == null) { // 如果未指定主键(通配符主键)，从原生message中取主键字段 - CanalEntry.Entry entry = flatMessage.getMessage().getEntries().get(0); + CanalEntry.Entry entry = flatMessage.getEntry(); CanalEntry.RowChange rowChange; try { rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index bf79ee45..d8957d92 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -3,7 +3,9 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; import java.util.ArrayList; import java.util.List; +import java.util.regex.Pattern; +import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; import com.alibaba.otter.canal.common.utils.CanalToStringStyle; @@ -89,4 +91,115 @@ public class Message implements Serializable { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } + /** + * 将 message 分区 + * + * @param partitionsNum 分区数 + * @param pkHashConfigs 分区库表主键正则表达式 + * @return 分区message数组 + */ + @SuppressWarnings("unchecked") + public Message[] messagePartition(Integer partitionsNum, String pkHashConfigs) { + if (partitionsNum == null) { + partitionsNum = 1; + } + Message[] partitionMessages = new Message[partitionsNum]; + List[] partitionEntries = new List[partitionsNum]; + for (int i = 0; i < partitionsNum; i++) { + partitionEntries[i] = new ArrayList<>(); + } + for (Entry entry : this.getEntries()) { + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException(e.getMessage(), e); + } + if (rowChange.getIsDdl()) { + partitionEntries[0].add(entry); + } else { + if (rowChange.getRowDatasList() != null && !rowChange.getRowDatasList().isEmpty()) { + String database = entry.getHeader().getSchemaName(); + String table = entry.getHeader().getTableName(); + + 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); + } + pkHashConfig = pkHashConfig.substring(0, i); + isMatch = Pattern.matches(pkHashConfig, database + "." + table); + if (isMatch) { + break; + } + } + + if (!isMatch) { + // 如果都没有匹配，发送到第一个分区 + partitionEntries[0].add(entry); + } else { + if (pk == null) { + // 如果未指定主键(通配符主键)，取主键字段 + 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) { + // 如果都没有匹配的主键，发送到第一个分区 + partitionEntries[0].add(entry); + } else { + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + boolean hasOldPk = false; + // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 + for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { + if (column.getName().equalsIgnoreCase(pk)) { + int hash = column.getValue().hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + hasOldPk = true; + break; + } + } + if (!hasOldPk) { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getName().equalsIgnoreCase(pk)) { + int hash = column.getValue().hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + break; + } + } + } + } + } + } + } + } + + for (int i = 0; i < partitionsNum; i++) { + List entries = partitionEntries[i]; + if (!entries.isEmpty()) { + partitionMessages[i] = new Message(this.id, entries); + } + } + + return partitionMessages; + } } 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 230e9607..0dd93ea5 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 @@ -80,22 +80,36 @@ public class CanalKafkaProducer implements CanalMQProducer { // producer.beginTransaction(); if (!kafkaProperties.getFlatMessage()) { try { - ProducerRecord record; + ProducerRecord record = null; if (canalDestination.getPartition() != null) { - record = new ProducerRecord(canalDestination.getTopic(), + record = new ProducerRecord<>(canalDestination.getTopic(), canalDestination.getPartition(), null, message); } else { - record = new ProducerRecord(canalDestination.getTopic(), 0, null, message); + if (canalDestination.getPartitionHash() != null && !canalDestination.getPartitionHash().isEmpty()) { + Message[] messages = message.messagePartition(canalDestination.getPartitionsNum(), + canalDestination.getPartitionHash()); + int length = messages.length; + for (int i = 0; i < length; i++) { + Message messagePartition = messages[i]; + if (messagePartition != null) { + record = new ProducerRecord<>(canalDestination.getTopic(), i, null, messagePartition); + } + } + } else { + record = new ProducerRecord<>(canalDestination.getTopic(), 0, null, message); + } } - producer.send(record).get(); + if (record != null) { + producer.send(record).get(); - if (logger.isDebugEnabled()) { - logger.debug("Send message to kafka topic: [{}], packet: {}", - canalDestination.getTopic(), - message.toString()); + if (logger.isDebugEnabled()) { + logger.debug("Send message to kafka topic: [{}], packet: {}", + canalDestination.getTopic(), + message.toString()); + } } } catch (Exception e) { logger.error(e.getMessage(), e); 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 74e3920b..56648d4e 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 @@ -1,14 +1,7 @@ package com.alibaba.otter.canal.rocketmq; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.serializer.SerializerFeature; -import com.alibaba.otter.canal.common.CanalMessageSerializer; -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 java.util.List; + import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; @@ -20,7 +13,15 @@ import org.apache.rocketmq.remoting.exception.RemotingException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.List; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; +import com.alibaba.otter.canal.common.CanalMessageSerializer; +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; public class CanalRocketMQProducer implements CanalMQProducer { @@ -34,8 +35,8 @@ public class CanalRocketMQProducer implements CanalMQProducer { public void init(MQProperties rocketMQProperties) { this.mqProperties = rocketMQProperties; RPCHook rpcHook = null; - if(rocketMQProperties.getAliyunAccessKey().length() > 0 - && rocketMQProperties.getAliyunSecretKey().length() > 0){ + if (rocketMQProperties.getAliyunAccessKey().length() > 0 + && rocketMQProperties.getAliyunSecretKey().length() > 0) { SessionCredentials sessionCredentials = new SessionCredentials(); sessionCredentials.setAccessKey(rocketMQProperties.getAliyunAccessKey()); sessionCredentials.setSecretKey(rocketMQProperties.getAliyunSecretKey()); @@ -59,23 +60,65 @@ public class CanalRocketMQProducer implements CanalMQProducer { Callback callback) { if (!mqProperties.getFlatMessage()) { try { - Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data, - mqProperties.isFilterTransactionEntry())); - logger.debug("send message:{} to destination:{}, partition: {}", - message, - destination.getCanalDestination(), - destination.getPartition()); - this.defaultMQProducer.send(message, new MessageQueueSelector() { - - @Override - public MessageQueue select(List mqs, Message msg, Object arg) { - int partition = 0; - if (destination.getPartition() != null) { - partition = destination.getPartition(); - } - return mqs.get(partition); + if (destination.getPartition() != null) { + Message message = new Message(destination.getTopic(), + CanalMessageSerializer.serializer(data, mqProperties.isFilterTransactionEntry())); + if (logger.isDebugEnabled()) { + logger.debug("send message:{} to destination:{}, partition: {}", + message, + destination.getCanalDestination(), + destination.getPartition()); } - }, null); + this.defaultMQProducer.send(message, new MessageQueueSelector() { + + @Override + public MessageQueue select(List mqs, Message msg, Object arg) { + int partition = 0; + if (destination.getPartition() != null) { + partition = destination.getPartition(); + } + return mqs.get(partition); + } + }, null); + } else { + if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { + com.alibaba.otter.canal.protocol.Message[] messages = data + .messagePartition(destination.getPartitionsNum(), destination.getPartitionHash()); + int length = messages.length; + for (int i = 0; i < length; i++) { + com.alibaba.otter.canal.protocol.Message dataPartition = messages[i]; + if (dataPartition != null) { + if (logger.isDebugEnabled()) { + logger.debug("flatMessagePart: {}, partition: {}", + JSON.toJSONString(dataPartition, SerializerFeature.WriteMapNullValue), + i); + } + final int index = i; + try { + Message message = new Message(destination.getTopic(), + CanalMessageSerializer.serializer(dataPartition, + mqProperties.isFilterTransactionEntry())); + this.defaultMQProducer.send(message, new MessageQueueSelector() { + + @Override + public MessageQueue select(List mqs, Message msg, Object arg) { + if (index > mqs.size()) { + throw new CanalServerException("partition number is error,config num:" + + destination.getPartitionsNum() + + ", mq num: " + mqs.size()); + } + return mqs.get(index); + } + }, null); + } catch (Exception e) { + logger.error("send flat message to hashed partition error", e); + callback.rollback(); + return; + } + } + } + } + } } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { logger.error("Send message error!", e); callback.rollback(); @@ -87,10 +130,12 @@ public class CanalRocketMQProducer implements CanalMQProducer { for (FlatMessage flatMessage : flatMessages) { if (destination.getPartition() != null) { try { - logger.info("send flat message: {} to topic: {} fixed partition: {}", - JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue), - destination.getTopic(), - destination.getPartition()); + if (logger.isDebugEnabled()) { + logger.debug("send message: {} to topic: {} fixed partition: {}", + JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue), + destination.getTopic(), + destination.getPartition()); + } Message message = new Message(destination.getTopic(), JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue).getBytes()); this.defaultMQProducer.send(message, new MessageQueueSelector() { @@ -114,13 +159,16 @@ public class CanalRocketMQProducer implements CanalMQProducer { for (int i = 0; i < length; i++) { FlatMessage flatMessagePart = partitionFlatMessage[i]; if (flatMessagePart != null) { - logger.debug("flatMessagePart: {}, partition: {}", - JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue), - i); + if (logger.isDebugEnabled()) { + logger.debug("flatMessagePart: {}, partition: {}", + JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue), + i); + } final int index = i; try { Message message = new Message(destination.getTopic(), - JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue).getBytes()); + JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue) + .getBytes()); this.defaultMQProducer.send(message, new MessageQueueSelector() { @Override From 5650b56bedc4eeb919a1f1389b5d67adf57be6c3 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 01:10:04 +0800 Subject: [PATCH 3/7] =?UTF-8?q?=E6=89=93=E5=8D=B0=E9=94=99=E8=AF=AF?= =?UTF-8?q?=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/launcher/loader/AbstractCanalAdapterWorker.java | 2 ++ 1 file changed, 2 insertions(+) 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 db87bf01..ec650231 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 @@ -70,6 +70,7 @@ public abstract class AbstractCanalAdapterWorker { }); return true; } catch (Exception e) { + logger.error(e.getMessage(), e); return false; } })); @@ -108,6 +109,7 @@ public abstract class AbstractCanalAdapterWorker { }); return true; } catch (Exception e) { + logger.error(e.getMessage(), e); return false; } })); From 99a1123e010eea0b087c11421932c820c34a6681 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 10 Dec 2018 09:52:48 +0800 Subject: [PATCH 4/7] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../resources/example/instance.properties | 6 ++-- .../alibaba/otter/canal/protocol/Message.java | 29 +++++++++++++++---- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 3a9ac883..8d884abd 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -1,5 +1,5 @@ ################################################# -## mysql serverId , v1.0.26+ will autoGen +## mysql serverId , v1.0.26+ will autoGen # canal.instance.mysql.slaveId=0 # enable gtid use true/false @@ -48,5 +48,5 @@ canal.mq.topic=example canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=3 -#canal.mq.partitionHash=mytest.person:id,mytest.role:id -################################################# \ No newline at end of file +#canal.mq.partitionHash=.*\\..*.$pk$ +################################################# diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index d8957d92..a01a8006 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -5,6 +5,7 @@ import java.util.ArrayList; import java.util.List; import java.util.regex.Pattern; +import com.google.protobuf.InvalidProtocolBufferException; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; @@ -93,7 +94,7 @@ public class Message implements Serializable { /** * 将 message 分区 - * + * * @param partitionsNum 分区数 * @param pkHashConfigs 分区库表主键正则表达式 * @return 分区message数组 @@ -108,7 +109,25 @@ public class Message implements Serializable { for (int i = 0; i < partitionsNum; i++) { partitionEntries[i] = new ArrayList<>(); } - for (Entry entry : this.getEntries()) { + + List entries; + if (this.isRaw()) { + List rawEntries = this.getRawEntries(); + entries = new ArrayList<>(rawEntries.size()); + for (ByteString byteString : rawEntries) { + Entry entry; + try { + entry = Entry.parseFrom(byteString); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + entries.add(entry); + } + } else { + entries = this.getEntries(); + } + + for (Entry entry : entries) { CanalEntry.RowChange rowChange; try { rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); @@ -194,9 +213,9 @@ public class Message implements Serializable { } for (int i = 0; i < partitionsNum; i++) { - List entries = partitionEntries[i]; - if (!entries.isEmpty()) { - partitionMessages[i] = new Message(this.id, entries); + List entriesTmp = partitionEntries[i]; + if (!entriesTmp.isEmpty()) { + partitionMessages[i] = new Message(this.id, entriesTmp); } } From 706ee88914c738073b0ae34180f10ce90fbb845b Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 10 Dec 2018 21:28:09 +0800 Subject: [PATCH 5/7] =?UTF-8?q?pattern=E8=A1=A8=E8=BE=BE=E5=BC=8F=E8=B0=83?= =?UTF-8?q?=E6=95=B4=20pk=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- protocol/pom.xml | 8 ++ .../otter/canal/protocol/FlatMessage.java | 78 ++++++----- .../alibaba/otter/canal/protocol/Message.java | 54 +++++--- .../protocol/aviater/AviaterRegexFilter.java | 124 ++++++++++++++++++ .../canal/protocol/aviater/PatternUtils.java | 38 ++++++ .../canal/protocol/aviater/RegexFunction.java | 31 +++++ 6 files changed, 283 insertions(+), 50 deletions(-) create mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java create mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java create mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java diff --git a/protocol/pom.xml b/protocol/pom.xml index 1b776b7e..0f7904d6 100644 --- a/protocol/pom.xml +++ b/protocol/pom.xml @@ -26,5 +26,13 @@ commons-lang commons-lang + + oro + oro + + + com.googlecode.aviator + aviator + 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 05d8f795..263b0d6e 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 @@ -2,8 +2,10 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; import java.util.*; -import java.util.regex.Pattern; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; import com.google.protobuf.ByteString; @@ -14,24 +16,27 @@ import com.google.protobuf.ByteString; */ public class FlatMessage implements Serializable { - private static final long serialVersionUID = -3386650678735860050L; + private static final long serialVersionUID = -3386650678735860050L; - private long id; - private String database; - private String table; - private Boolean isDdl; - private String type; + private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); + private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); + + private long id; + private String database; + private String table; + private Boolean isDdl; + private String type; // binlog executeTime - private Long es; + private Long es; // dml build timeStamp - private Long ts; - private String sql; - private Map sqlType; - private Map mysqlType; - private List> data; - private List> old; + private Long ts; + private String sql; + private Map sqlType; + private Map mysqlType; + private List> data; + private List> old; - private transient CanalEntry.Entry entry; // 所属entry + private transient CanalEntry.Entry entry; // 所属entry public FlatMessage(long id){ this.id = id; @@ -143,7 +148,7 @@ public class FlatMessage implements Serializable { /** * 将Message转换为FlatMessage - * + * * @param message 原生message * @return FlatMessage列表 */ @@ -272,7 +277,7 @@ public class FlatMessage implements Serializable { /** * 将FlatMessage按指定的字段值hash拆分 - * + * * @param flatMessage flatMessage * @param partitionsNum 分区数量 * @param pkHashConfigs hash映射 @@ -302,7 +307,14 @@ public class FlatMessage implements Serializable { pk = pkHashConfig.substring(i + 1); } pkHashConfig = pkHashConfig.substring(0, i); - isMatch = Pattern.matches(pkHashConfig, database + "." + table); + + AviaterRegexFilter aviaterRegexFilter = regexFilters.get(pkHashConfig); + if (aviaterRegexFilter == null) { + aviaterRegexFilter = new AviaterRegexFilter(pkHashConfig); + regexFilters.putIfAbsent(pkHashConfig, aviaterRegexFilter); + } + + isMatch = aviaterRegexFilter.filter(database + "." + table); if (isMatch) { break; } @@ -313,19 +325,23 @@ public class FlatMessage implements Serializable { partitionMessages[0] = flatMessage; } else { if (pk == null) { - // 如果未指定主键(通配符主键)，从原生message中取主键字段 - CanalEntry.Entry entry = flatMessage.getEntry(); - 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; + pk = schemaTabPk.get(database + "." + table); + if (pk == null) { + // 如果未指定主键(通配符主键)，从原生message中取主键字段 + CanalEntry.Entry entry = flatMessage.getEntry(); + 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(); + schemaTabPk.putIfAbsent(database + "." + table, pk); + break; + } } } } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index a01a8006..00837fec 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -3,15 +3,17 @@ package com.alibaba.otter.canal.protocol; import java.io.Serializable; import java.util.ArrayList; import java.util.List; -import java.util.regex.Pattern; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; -import com.google.protobuf.InvalidProtocolBufferException; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; import com.alibaba.otter.canal.common.utils.CanalToStringStyle; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.google.protobuf.ByteString; +import com.google.protobuf.InvalidProtocolBufferException; /** * @author zebin.xuzb @ 2012-6-19 @@ -19,14 +21,17 @@ import com.google.protobuf.ByteString; */ public class Message implements Serializable { - private static final long serialVersionUID = 1234034768477580009L; + private static final long serialVersionUID = 1234034768477580009L; - private long id; - private List entries = new ArrayList(); + private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); + private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); + + private long id; + private List entries = new ArrayList(); // row data for performance, see: // https://github.com/alibaba/canal/issues/726 - private boolean raw = true; - private List rawEntries = new ArrayList(); + private boolean raw = true; + private List rawEntries = new ArrayList(); public Message(long id, List entries){ this.id = id; @@ -152,7 +157,14 @@ public class Message implements Serializable { pk = pkHashConfig.substring(i + 1); } pkHashConfig = pkHashConfig.substring(0, i); - isMatch = Pattern.matches(pkHashConfig, database + "." + table); + + AviaterRegexFilter aviaterRegexFilter = regexFilters.get(pkHashConfig); + if (aviaterRegexFilter == null) { + aviaterRegexFilter = new AviaterRegexFilter(pkHashConfig); + regexFilters.putIfAbsent(pkHashConfig, aviaterRegexFilter); + } + + isMatch = aviaterRegexFilter.filter(database + "." + table); if (isMatch) { break; } @@ -163,17 +175,21 @@ public class Message implements Serializable { partitionEntries[0].add(entry); } else { if (pk == null) { - // 如果未指定主键(通配符主键)，取主键字段 - 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; + pk = schemaTabPk.get(database + "." + table); + if (pk == null) { + // 如果未指定主键(通配符主键)，取主键字段 + 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(); + schemaTabPk.putIfAbsent(database + "." + table, pk); + break; + } } } } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java new file mode 100644 index 00000000..db7750ba --- /dev/null +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/AviaterRegexFilter.java @@ -0,0 +1,124 @@ +package com.alibaba.otter.canal.protocol.aviater; + +import java.util.*; + +import org.apache.commons.lang.StringUtils; + +import com.googlecode.aviator.AviatorEvaluator; +import com.googlecode.aviator.Expression; + +/** + * 基于aviater进行tableName正则匹配的过滤算法 + * + * @author jianghang 2012-7-20 下午06:01:34 + */ +public class AviaterRegexFilter { + + private static final String SPLIT = ","; + private static final String PATTERN_SPLIT = "|"; + private static final String FILTER_EXPRESSION = "regex(pattern,target)"; + private static final RegexFunction regexFunction = new RegexFunction(); + private final Expression exp = AviatorEvaluator.compile(FILTER_EXPRESSION, true); + static { + AviatorEvaluator.addFunction(regexFunction); + } + + private static final Comparator COMPARATOR = new StringComparator(); + + final private String pattern; + final private boolean defaultEmptyValue; + + public AviaterRegexFilter(String pattern){ + this(pattern, true); + } + + public AviaterRegexFilter(String pattern, boolean defaultEmptyValue){ + this.defaultEmptyValue = defaultEmptyValue; + List list = null; + if (StringUtils.isEmpty(pattern)) { + list = new ArrayList(); + } else { + String[] ss = StringUtils.split(pattern, SPLIT); + list = Arrays.asList(ss); + } + + // 对pattern按照从长到短的排序 + // 因为 foo|foot 匹配 foot 会出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot + // 的长度不一样 + Collections.sort(list, COMPARATOR); + // 对pattern进行头尾完全匹配 + list = completionPattern(list); + this.pattern = StringUtils.join(list, PATTERN_SPLIT); + } + + public boolean filter(String filtered) { + if (StringUtils.isEmpty(pattern)) { + return defaultEmptyValue; + } + + if (StringUtils.isEmpty(filtered)) { + return defaultEmptyValue; + } + + Map env = new HashMap(); + env.put("pattern", pattern); + env.put("target", filtered.toLowerCase()); + return (Boolean) exp.execute(env); + } + + /** + * 修复正则表达式匹配的问题，因为使用了 oro 的 matches，会出现： + * + * 
+     * foo|foot 匹配 foot 出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot 的长度不一样
+     *
+ * + * 因此此类对正则表达式进行了从长到短的排序 + * + * @author zebin.xuzb 2012-10-22 下午2:02:26 + * @version 1.0.0 + */ + private static class StringComparator implements Comparator { + + @Override + public int compare(String str1, String str2) { + if (str1.length() > str2.length()) { + return -1; + } else if (str1.length() < str2.length()) { + return 1; + } else { + return 0; + } + } + } + + /** + * 修复正则表达式匹配的问题，即使按照长度递减排序，还是会出现以下问题： + * + * 
+     * foooo|f.*t 匹配 fooooot 出错，原因是 fooooot 匹配了 foooo 之后，会将 fooo 和数据进行匹配，但是 foooo 的长度和 fooooot 的长度不一样
+     *
+ * + * 因此此类对正则表达式进行头尾完全匹配 + * + * @author simon + * @version 1.0.0 + */ + + private List completionPattern(List patterns) { + List result = new ArrayList(); + for (String pattern : patterns) { + StringBuffer stringBuffer = new StringBuffer(); + stringBuffer.append("^"); + stringBuffer.append(pattern); + stringBuffer.append("$"); + result.add(stringBuffer.toString()); + } + return result; + } + + @Override + public String toString() { + return pattern; + } +} diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java new file mode 100644 index 00000000..e04f0c4b --- /dev/null +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/PatternUtils.java @@ -0,0 +1,38 @@ +package com.alibaba.otter.canal.protocol.aviater; + +import java.util.Map; + +import org.apache.oro.text.regex.MalformedPatternException; +import org.apache.oro.text.regex.Pattern; +import org.apache.oro.text.regex.PatternCompiler; +import org.apache.oro.text.regex.Perl5Compiler; + +import com.google.common.base.Function; +import com.google.common.collect.MapMaker; +import com.google.common.collect.MigrateMap; + +public class PatternUtils { + @SuppressWarnings("deprecation") + private static Map patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(), + new Function() { + + public Pattern apply(String pattern) { + try { + PatternCompiler pc = new Perl5Compiler(); + return pc.compile(pattern, + Perl5Compiler.CASE_INSENSITIVE_MASK | Perl5Compiler.READ_ONLY_MASK + | Perl5Compiler.SINGLELINE_MASK); + } catch (MalformedPatternException e) { + throw new RuntimeException(e); + } + } + }); + + public static Pattern getPattern(String pattern) { + return patterns.get(pattern); + } + + public static void clear() { + patterns.clear(); + } +} diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java new file mode 100644 index 00000000..be084c25 --- /dev/null +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/aviater/RegexFunction.java @@ -0,0 +1,31 @@ +package com.alibaba.otter.canal.protocol.aviater; + +import java.util.Map; + +import org.apache.oro.text.regex.Perl5Matcher; + +import com.googlecode.aviator.runtime.function.AbstractFunction; +import com.googlecode.aviator.runtime.function.FunctionUtils; +import com.googlecode.aviator.runtime.type.AviatorBoolean; +import com.googlecode.aviator.runtime.type.AviatorObject; + +/** + * 提供aviator regex的代码扩展 + * + * @author jianghang 2012-7-23 上午10:29:23 + */ +public class RegexFunction extends AbstractFunction { + + public AviatorObject call(Map env, AviatorObject arg1, AviatorObject arg2) { + String pattern = FunctionUtils.getStringValue(arg1, env); + String text = FunctionUtils.getStringValue(arg2, env); + Perl5Matcher matcher = new Perl5Matcher(); + boolean isMatch = matcher.matches(text, PatternUtils.getPattern(pattern)); + return AviatorBoolean.valueOf(isMatch); + } + + public String getName() { + return "regex"; + } + +} From eac3f5ce5fa2152136e525e1909c91d7373381f8 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 23:54:07 +0800 Subject: [PATCH 6/7] =?UTF-8?q?=E4=BF=AE=E6=94=B9convert=E6=97=B6=E8=AE=BE?= =?UTF-8?q?=E7=BD=AEpk?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/protocol/FlatMessage.java | 35 ++++----- .../alibaba/otter/canal/protocol/Message.java | 77 +++++++++---------- 2 files changed, 49 insertions(+), 63 deletions(-) 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 263b0d6e..53924d4a 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 @@ -5,9 +5,9 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import com.google.protobuf.ByteString; /** @@ -18,12 +18,12 @@ public class FlatMessage implements Serializable { private static final long serialVersionUID = -3386650678735860050L; - private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); private long id; private String database; private String table; + private String pk; private Boolean isDdl; private String type; // binlog executeTime @@ -66,6 +66,14 @@ public class FlatMessage implements Serializable { this.table = table; } + public String getPk() { + return pk; + } + + public void setPk(String pk) { + this.pk = pk; + } + public Boolean getIsDdl() { return isDdl; } @@ -222,6 +230,9 @@ public class FlatMessage implements Serializable { } for (CanalEntry.Column column : columns) { + if (flatMessage.getPk() == null && column.getIsKey()) { + flatMessage.setPk(column.getName()); + } sqlType.put(column.getName(), column.getSqlType()); mysqlType.put(column.getName(), column.getMysqlType()); if (column.getIsNull()) { @@ -325,25 +336,7 @@ public class FlatMessage implements Serializable { partitionMessages[0] = flatMessage; } else { if (pk == null) { - pk = schemaTabPk.get(database + "." + table); - if (pk == null) { - // 如果未指定主键(通配符主键)，从原生message中取主键字段 - CanalEntry.Entry entry = flatMessage.getEntry(); - 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(); - schemaTabPk.putIfAbsent(database + "." + table, pk); - break; - } - } - } + pk = flatMessage.getPk(); } if (pk == null || !flatMessage.getData().get(0).containsKey(pk)) { // 如果都没有匹配的主键，发送到第一个分区 diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 00837fec..6fc42a41 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -4,14 +4,13 @@ import java.io.Serializable; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; import com.alibaba.otter.canal.common.utils.CanalToStringStyle; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; +import com.alibaba.otter.canal.protocol.aviater.AviaterRegexFilter; import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; @@ -23,7 +22,6 @@ public class Message implements Serializable { private static final long serialVersionUID = 1234034768477580009L; - private static ConcurrentMap schemaTabPk = new ConcurrentHashMap<>(); private static ConcurrentHashMap regexFilters = new ConcurrentHashMap<>(); private long id; @@ -173,55 +171,50 @@ public class Message implements Serializable { if (!isMatch) { // 如果都没有匹配，发送到第一个分区 partitionEntries[0].add(entry); - } else { - if (pk == null) { - pk = schemaTabPk.get(database + "." + table); - if (pk == null) { - // 如果未指定主键(通配符主键)，取主键字段 - try { - rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - } catch (Exception e) { - throw new RuntimeException(e.getMessage(), e); + } + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + boolean hasOldPk = false; + // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 + for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { + CanalEntry.Column pkColumn = null; + if (pk != null) { + if (column.getName().equalsIgnoreCase(pk)) { + pkColumn = column; } - CanalEntry.RowData rowData = rowChange.getRowDatasList().get(0); - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getIsKey()) { - pk = column.getName(); - schemaTabPk.putIfAbsent(database + "." + table, pk); - break; - } + } else { + if (column.getIsKey()) { + pkColumn = column; } } + if (pkColumn != null) { + int hash = pkColumn.getValue().hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + hasOldPk = true; + break; + } } - } - if (pk == null) { - // 如果都没有匹配的主键，发送到第一个分区 - partitionEntries[0].add(entry); - } else { - for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - boolean hasOldPk = false; - // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 - for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { - if (column.getName().equalsIgnoreCase(pk)) { - int hash = column.getValue().hashCode(); + if (!hasOldPk) { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + CanalEntry.Column pkColumn = null; + if (pk != null) { + if (column.getName().equalsIgnoreCase(pk)) { + pkColumn = column; + } + } else { + if (column.getIsKey()) { + pkColumn = column; + } + } + if (pkColumn != null) { + int hash = pkColumn.getValue().hashCode(); int pkHash = Math.abs(hash) % partitionsNum; pkHash = Math.abs(pkHash); partitionEntries[pkHash].add(entry); - hasOldPk = true; break; } } - if (!hasOldPk) { - for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - if (column.getName().equalsIgnoreCase(pk)) { - int hash = column.getValue().hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - pkHash = Math.abs(pkHash); - partitionEntries[pkHash].add(entry); - break; - } - } - } } } } From aeecf87490eee5dc297dfea7d3b5f25c660d8a80 Mon Sep 17 00:00:00 2001 From: machey Date: Tue, 11 Dec 2018 00:09:12 +0800 Subject: [PATCH 7/7] =?UTF-8?q?=E5=8E=BB=E6=8E=89=E5=AF=B9before=E7=9A=84?= =?UTF-8?q?=E5=88=A4=E6=96=AD=EF=BC=8C=E5=88=86=E5=8C=BA=E5=8F=91=E9=80=81?= =?UTF-8?q?=E7=9A=84=E8=A1=A8pk=E4=B8=8D=E8=83=BD=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/protocol/FlatMessage.java | 12 +--- .../alibaba/otter/canal/protocol/Message.java | 58 +++++++------------ 2 files changed, 21 insertions(+), 49 deletions(-) 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 53924d4a..f46c26a4 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 @@ -344,17 +344,7 @@ public class FlatMessage implements Serializable { } 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); - } + String value = row.get(pk); if (value == null) { value = ""; } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 6fc42a41..1013bfe6 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -173,48 +173,30 @@ public class Message implements Serializable { partitionEntries[0].add(entry); } for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { - boolean hasOldPk = false; - // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 - for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { - CanalEntry.Column pkColumn = null; - if (pk != null) { - if (column.getName().equalsIgnoreCase(pk)) { - pkColumn = column; - } - } else { - if (column.getIsKey()) { - pkColumn = column; - } - } - if (pkColumn != null) { - int hash = pkColumn.getValue().hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - pkHash = Math.abs(pkHash); - partitionEntries[pkHash].add(entry); - hasOldPk = true; - break; - } - } - if (!hasOldPk) { + String pkValue = null; + if (pk != null) { for (CanalEntry.Column column : rowData.getAfterColumnsList()) { - CanalEntry.Column pkColumn = null; - if (pk != null) { - if (column.getName().equalsIgnoreCase(pk)) { - pkColumn = column; - } - } else { - if (column.getIsKey()) { - pkColumn = column; - } - } - if (pkColumn != null) { - int hash = pkColumn.getValue().hashCode(); - int pkHash = Math.abs(hash) % partitionsNum; - pkHash = Math.abs(pkHash); - partitionEntries[pkHash].add(entry); + if (column.getName().equalsIgnoreCase(pk)) { + pkValue = column.getValue(); break; } } + } else { + for (CanalEntry.Column column : rowData.getAfterColumnsList()) { + if (column.getIsKey()) { + pkValue = column.getValue(); + break; + } + } + } + if (pkValue != null) { + int hash = pkValue.hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + pkHash = Math.abs(pkHash); + partitionEntries[pkHash].add(entry); + } else { + // 没有找到主键 + partitionEntries[0].add(entry); } } }