chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

去掉对before的判断，分区发送的表pk不能修改

Browse Source
This commit is contained in:
machey
2018-12-11 00:09:12 +08:00
parent eac3f5ce5f
commit aeecf87490
2 changed files with 21 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
View File

@@ -344,17 +344,7 @@ public class FlatMessage implements Serializable {
} else { } else {
int idx = 0; int idx = 0;
for (Map<String, String> row : flatMessage.getData()) { for (Map<String, String> row : flatMessage.getData()) {
Map<String, String> o = null; String value = row.get(pk);
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) { if (value == null) {
value = ""; value = "";
} }

58
protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java
View File

@@ -173,48 +173,30 @@ public class Message implements Serializable {
partitionEntries[0].add(entry); partitionEntries[0].add(entry);
} }
for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
boolean hasOldPk = false; String pkValue = null;
// 如果before中有pk值说明主键有修改, 以旧的主键值hash为准 if (pk != null) {
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) {
for (CanalEntry.Column column : rowData.getAfterColumnsList()) { for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
CanalEntry.Column pkColumn = null; if (column.getName().equalsIgnoreCase(pk)) {
if (pk != null) { pkValue = column.getValue();
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);
break; 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);
} }
} }
} }