From eac3f5ce5fa2152136e525e1909c91d7373381f8 Mon Sep 17 00:00:00 2001 From: machey Date: Mon, 10 Dec 2018 23:54:07 +0800 Subject: [PATCH] =?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; - } - } - } } } }