From 92f04f86cb0956fc958012baa336ed02b89e486d Mon Sep 17 00:00:00 2001 From: 844392610 <844392610@qq.com> Date: Fri, 28 Feb 2020 12:22:17 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E7=9B=AE=E6=A0=87=E8=A1=A8?= =?UTF-8?q?=E4=B8=8D=E5=8C=BA=E5=88=86=E5=A4=A7=E5=B0=8F=E5=86=99=E7=9A=84?= =?UTF-8?q?=E6=94=AF=E6=8C=81=EF=BC=8C=E5=B9=B6=E5=8F=AF=E5=BC=80=E5=85=B3?= =?UTF-8?q?=EF=BC=88=E9=BB=98=E8=AE=A4=E4=B8=BA=E5=8C=BA=E5=88=86=E5=A4=A7?= =?UTF-8?q?=E5=B0=8F=E5=86=99=EF=BC=89=20(#2329)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 添加目标表不区分大小写的支持，并可开关（默认为区分大小写） * 添加目标表不区分大小写的支持，并可开关（默认为区分大小写）-format --- .../adapter/rdb/config/MappingConfig.java | 32 ++++--- .../rdb/service/RdbMirrorDbSyncService.java | 7 +- .../adapter/rdb/service/RdbSyncService.java | 84 +++++++++---------- .../client/adapter/rdb/support/SingleDml.java | 25 +++++- 4 files changed, 89 insertions(+), 59 deletions(-) diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java index 1b0276dd..294f2735 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfig.java @@ -92,19 +92,21 @@ public class MappingConfig implements AdapterConfig { public static class DbMapping implements AdapterMapping { - private boolean mirrorDb = false; // 是否镜像库 - private String database; // 数据库名或schema名 - private String table; // 表名 - private Map targetPk = new LinkedHashMap<>(); // 目标表主键字段 - private boolean mapAll = false; // 映射所有字段 - private String targetDb; // 目标库名 - private String targetTable; // 目标表名 - private Map targetColumns; // 目标表字段映射 + private boolean mirrorDb = false; // 是否镜像库 + private String database; // 数据库名或schema名 + private String table; // 表名 + private Map targetPk = new LinkedHashMap<>(); // 目标表主键字段 + private boolean mapAll = false; // 映射所有字段 + private String targetDb; // 目标库名 + private String targetTable; // 目标表名 + private Map targetColumns; // 目标表字段映射 - private String etlCondition; // etl条件sql + private boolean caseInsensitive = false; // 目标表不区分大小写，默认是否 - private int readBatch = 5000; - private int commitBatch = 5000; // etl等批量提交大小 + private String etlCondition; // etl条件sql + + private int readBatch = 5000; + private int commitBatch = 5000; // etl等批量提交大小 private Map allMapColumns; @@ -179,6 +181,14 @@ public class MappingConfig implements AdapterConfig { this.targetColumns = targetColumns; } + public boolean isCaseInsensitive() { + return caseInsensitive; + } + + public void setCaseInsensitive(boolean caseInsensitive) { + this.caseInsensitive = caseInsensitive; + } + public String getEtlCondition() { return etlCondition; } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java index 9d574fc6..52862337 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbMirrorDbSyncService.java @@ -91,9 +91,10 @@ public class RdbMirrorDbSyncService { if (config == null) { return false; } - + // 是否区分大小写 + boolean caseInsensitive = config.getDbMapping().isCaseInsensitive(); if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); singleDmls.forEach(singleDml -> { int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData()); RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); @@ -101,7 +102,7 @@ public class RdbMirrorDbSyncService { }); } else { int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); singleDmls.forEach(singleDml -> { RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); rdbSyncService.getDmlsPartition()[hash].add(syncItem); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 0780c4f0..73392751 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -112,9 +112,8 @@ public class RdbSyncService { futures.add(executorThreads[i].submit(() -> { try { - dmlsPartition[j].forEach(syncItem -> sync(batchExecutors[j], - syncItem.config, - syncItem.singleDml)); + dmlsPartition[j] + .forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml)); dmlsPartition[j].clear(); batchExecutors[j].commit(); return true; @@ -152,49 +151,50 @@ public class RdbSyncService { sync(dmls, dml -> { if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { // DDL - columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); - return false; - } else { - // DML - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String groupId = StringUtils.trimToEmpty(dml.getGroupId()); - String database = dml.getDatabase(); - String table = dml.getTable(); - Map configMap; - if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { - configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table); + columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); + return false; } else { - configMap = mappingConfig.get(destination + "_" + database + "-" + table); - } - - if (configMap == null) { - return false; - } - - if (configMap.values().isEmpty()) { - return false; - } - - for (MappingConfig config : configMap.values()) { - if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - int hash = pkHash(config.getDbMapping(), singleDml.getData()); - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); + // DML + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String groupId = StringUtils.trimToEmpty(dml.getGroupId()); + String database = dml.getDatabase(); + String table = dml.getTable(); + Map configMap; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table); } else { - int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); + configMap = mappingConfig.get(destination + "_" + database + "-" + table); } + + if (configMap == null) { + return false; + } + + if (configMap.values().isEmpty()) { + return false; + } + + for (MappingConfig config : configMap.values()) { + boolean caseInsensitive = config.getDbMapping().isCaseInsensitive(); + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); + singleDmls.forEach(singleDml -> { + int hash = pkHash(config.getDbMapping(), singleDml.getData()); + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml, caseInsensitive); + singleDmls.forEach(singleDml -> { + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } + } + return true; } - return true; - } - } ); + }); } /** diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java index bad12249..126ad0ea 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java @@ -5,6 +5,7 @@ import java.util.List; import java.util.Map; import com.alibaba.otter.canal.client.adapter.support.Dml; +import org.springframework.util.LinkedCaseInsensitiveMap; public class SingleDml { @@ -63,7 +64,7 @@ public class SingleDml { this.old = old; } - public static List dml2SingleDmls(Dml dml) { + public static List dml2SingleDmls(Dml dml, boolean caseInsensitive) { List singleDmls = new ArrayList<>(); if (dml.getData() != null) { int size = dml.getData().size(); @@ -73,9 +74,17 @@ public class SingleDml { singleDml.setDatabase(dml.getDatabase()); singleDml.setTable(dml.getTable()); singleDml.setType(dml.getType()); - singleDml.setData(dml.getData().get(i)); + Map data = dml.getData().get(i); + if (caseInsensitive) { + data = toCaseInsensitiveMap(data); + } + singleDml.setData(data); if (dml.getOld() != null) { - singleDml.setOld(dml.getOld().get(i)); + Map oldData = dml.getOld().get(i); + if (caseInsensitive) { + oldData = toCaseInsensitiveMap(oldData); + } + singleDml.setOld(oldData); } singleDmls.add(singleDml); } @@ -89,4 +98,14 @@ public class SingleDml { } return singleDmls; } + + public static List dml2SingleDmls(Dml dml) { + return dml2SingleDmls(dml, false); + } + + private static LinkedCaseInsensitiveMap toCaseInsensitiveMap(Map data) { + LinkedCaseInsensitiveMap map = new LinkedCaseInsensitiveMap(); + map.putAll(data); + return map; + } }