diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java index 8d49e726..132199fc 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java @@ -67,11 +67,11 @@ public class YmlConfigBinder { } /** - * 将当前内容指定前缀部分绑定到指定对象并用环境变量中的属性替换占位符, 例: - * 当前内容有属性 zkServers: ${zookeeper.servers} - * 在envProperties中有属性 zookeeper.servers: 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 - * 则当前内容 zkServers 会被替换为 zkServers: 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 - * 注: 假设绑定的类中 zkServers 属性是 List 对象, 则会自动映射成List + * 将当前内容指定前缀部分绑定到指定对象并用环境变量中的属性替换占位符, 例: 当前内容有属性 zkServers: ${zookeeper.servers} + * 在envProperties中有属性 zookeeper.servers: + * 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 则当前内容 zkServers 会被替换为 + * zkServers: 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 注: 假设绑定的类中 + * zkServers 属性是 List 对象, 则会自动映射成List * * @param prefix 指定前缀 * @param content yml内容 @@ -92,6 +92,10 @@ public class YmlConfigBinder { Resource configResource = new ByteArrayResource(contentBytes); PropertySource propertySource = propertySourceLoader.load("manualBindConfig", configResource, null); + if (propertySource == null) { + return null; + } + Properties properties = new Properties(); Map propertiesRes = new LinkedHashMap<>(); if (!StringUtils.isEmpty(prefix) && !prefix.endsWith(".")) { @@ -126,6 +130,10 @@ public class YmlConfigBinder { propertiesRes.put(key, value); } + if (propertiesRes.isEmpty()) { + return null; + } + propertySource = new MapPropertySource(propertySource.getName(), propertiesRes); T target = clazz.newInstance(); diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java index 239b8839..93ca987b 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java @@ -43,6 +43,7 @@ public class MessageUtil { dml.setTable(entry.getHeader().getTableName()); dml.setType(eventType.toString()); dml.setEs(entry.getHeader().getExecuteTime()); + dml.setIsDdl(rowChange.getIsDdl()); dml.setTs(System.currentTimeMillis()); dml.setSql(rowChange.getSql()); dmls.add(dml); diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 9e54cce3..34b7a38a 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -1,7 +1,10 @@ package com.alibaba.otter.canal.client.adapter.es; import java.net.InetAddress; -import java.util.*; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; import java.util.concurrent.ConcurrentHashMap; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -46,6 +49,8 @@ public class ESAdapter implements OuterAdapter { private ESConfigMonitor esConfigMonitor; + private Properties envProperties; + public TransportClient getTransportClient() { return transportClient; } @@ -65,6 +70,7 @@ public class ESAdapter implements OuterAdapter { @Override public void init(OuterAdapterConfig configuration, Properties envProperties) { try { + this.envProperties = envProperties; Map esSyncConfigTmp = ESSyncConfigLoader.load(envProperties); // 过滤不匹配的key的配置 esSyncConfigTmp.forEach((key, config) -> { @@ -93,10 +99,21 @@ public class ESAdapter implements OuterAdapter { String schema = matcher.group(2); schemaItem.getAliasTableItems().values().forEach(tableItem -> { - Map esSyncConfigMap = dbTableEsSyncConfig.computeIfAbsent( - StringUtils.trimToEmpty(config.getDestination()) + "-" + StringUtils - .trimToEmpty(config.getGroupId()) + "_" + schema + "-" + tableItem.getTableName(), - k -> new ConcurrentHashMap<>()); + Map esSyncConfigMap; + if (envProperties != null + && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + esSyncConfigMap = dbTableEsSyncConfig + .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "-" + + StringUtils.trimToEmpty(config.getGroupId()) + "_" + schema + "-" + + tableItem.getTableName(), + k -> new ConcurrentHashMap<>()); + } else { + esSyncConfigMap = dbTableEsSyncConfig + .computeIfAbsent(StringUtils.trimToEmpty(config.getDestination()) + "_" + schema + "-" + + tableItem.getTableName(), + k -> new ConcurrentHashMap<>()); + } + esSyncConfigMap.put(configName, config); }); } @@ -134,9 +151,16 @@ public class ESAdapter implements OuterAdapter { public void sync(Dml dml) { String database = dml.getDatabase(); String table = dml.getTable(); - Map configMap = dbTableEsSyncConfig - .get(StringUtils.trimToEmpty(dml.getDestination()) + "-" + StringUtils.trimToEmpty(dml.getGroupId()) + "_" - + database + "-" + table); + Map configMap; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + configMap = dbTableEsSyncConfig + .get(StringUtils.trimToEmpty(dml.getDestination()) + "-" + StringUtils.trimToEmpty(dml.getGroupId()) + + "_" + database + "-" + table); + } else { + configMap = dbTableEsSyncConfig + .get(StringUtils.trimToEmpty(dml.getDestination()) + "_" + database + "-" + table); + } + if (configMap != null && !configMap.values().isEmpty()) { esSyncService.sync(configMap.values(), dml); } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java index d721cc3d..51216665 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java @@ -28,7 +28,9 @@ public class ESSyncConfigLoader { Map configContentMap = MappingConfigsLoader.loadConfigs("es"); configContentMap.forEach((fileName, content) -> { ESSyncConfig config = YmlConfigBinder.bindYmlToObj(null, content, ESSyncConfig.class, null, envProperties); - + if (config == null) { + return; + } try { config.validate(); } catch (Exception e) { diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index 795631e0..c307aec9 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -46,6 +46,8 @@ public class HbaseAdapter implements OuterAdapter { private HbaseConfigMonitor configMonitor; + private Properties envProperties; + public Map getHbaseMapping() { return hbaseMapping; } @@ -57,6 +59,7 @@ public class HbaseAdapter implements OuterAdapter { @Override public void init(OuterAdapterConfig configuration, Properties envProperties) { try { + this.envProperties = envProperties; Map hbaseMappingTmp = MappingConfigLoader.load(envProperties); // 过滤不匹配的key的配置 hbaseMappingTmp.forEach((key, mappingConfig) -> { @@ -69,11 +72,19 @@ public class HbaseAdapter implements OuterAdapter { for (Map.Entry entry : hbaseMapping.entrySet()) { String configName = entry.getKey(); MappingConfig mappingConfig = entry.getValue(); - String k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-" - + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_" - + mappingConfig.getHbaseMapping().getDatabase() + "-" - + mappingConfig.getHbaseMapping().getTable(); - Map configMap = mappingConfigCache.computeIfAbsent(k, k1 -> new ConcurrentHashMap<>()); + String k; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-" + + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_" + + mappingConfig.getHbaseMapping().getDatabase() + "-" + + mappingConfig.getHbaseMapping().getTable(); + } else { + k = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_" + + mappingConfig.getHbaseMapping().getDatabase() + "-" + + mappingConfig.getHbaseMapping().getTable(); + } + Map configMap = mappingConfigCache.computeIfAbsent(k, + k1 -> new ConcurrentHashMap<>()); configMap.put(configName, mappingConfig); } @@ -108,8 +119,12 @@ public class HbaseAdapter implements OuterAdapter { String groupId = StringUtils.trimToEmpty(dml.getGroupId()); String database = dml.getDatabase(); String table = dml.getTable(); - Map configMap = mappingConfigCache - .get(destination + "-" + groupId + "_" + database + "-" + table); + Map configMap; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + configMap = mappingConfigCache.get(destination + "-" + groupId + "_" + database + "-" + table); + } else { + configMap = mappingConfigCache.get(destination + "_" + database + "-" + table); + } if (configMap != null) { List configs = new ArrayList<>(); configMap.values().forEach(config -> { diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java index ff81e6b7..2b43492f 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java @@ -34,6 +34,9 @@ public class MappingConfigLoader { configContentMap.forEach((fileName, content) -> { MappingConfig config = YmlConfigBinder .bindYmlToObj(null, content, MappingConfig.class, null, envProperties); + if (config == null) { + return; + } try { config.validate(); } catch (Exception e) { diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index d06e03c1..8d1cab75 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -49,6 +49,8 @@ public class RdbAdapter implements OuterAdapter { private RdbConfigMonitor rdbConfigMonitor; + private Properties envProperties; + public Map getRdbMapping() { return rdbMapping; } @@ -68,6 +70,7 @@ public class RdbAdapter implements OuterAdapter { */ @Override public void init(OuterAdapterConfig configuration, Properties envProperties) { + this.envProperties = envProperties; Map rdbMappingTmp = ConfigLoader.load(envProperties); // 过滤不匹配的key的配置 rdbMappingTmp.forEach((key, mappingConfig) -> { @@ -86,10 +89,15 @@ public class RdbAdapter implements OuterAdapter { String configName = entry.getKey(); MappingConfig mappingConfig = entry.getValue(); if (!mappingConfig.getDbMapping().getMirrorDb()) { - String key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-" - + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_" - + mappingConfig.getDbMapping().getDatabase() + "-" - + mappingConfig.getDbMapping().getTable(); + String key; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "-" + + StringUtils.trimToEmpty(mappingConfig.getGroupId()) + "_" + + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable(); + } else { + key = StringUtils.trimToEmpty(mappingConfig.getDestination()) + "_" + + mappingConfig.getDbMapping().getDatabase() + "-" + mappingConfig.getDbMapping().getTable(); + } Map configMap = mappingConfigCache.computeIfAbsent(key, k1 -> new ConcurrentHashMap<>()); configMap.put(configName, mappingConfig); @@ -152,7 +160,7 @@ public class RdbAdapter implements OuterAdapter { return; } try { - rdbSyncService.sync(mappingConfigCache, dmls); + rdbSyncService.sync(mappingConfigCache, dmls, envProperties); rdbMirrorDbSyncService.sync(dmls); } catch (Exception e) { throw new RuntimeException(e); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java index 63dd728a..514d9a79 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java @@ -36,6 +36,9 @@ public class ConfigLoader { configContentMap.forEach((fileName, content) -> { MappingConfig config = YmlConfigBinder .bindYmlToObj(null, content, MappingConfig.class, null, envProperties); + if (config == null) { + return; + } try { config.validate(); } catch (Exception e) { 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 eec96e0c..9d574fc6 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 @@ -59,7 +59,7 @@ public class RdbMirrorDbSyncService { if (mirrorDbConfig.getMappingConfig() == null) { continue; } - if (StringUtils.isNotEmpty(mirrorDbConfig.getMappingConfig().getGroupId())) { + if (dml.getGroupId() != null && StringUtils.isNotEmpty(mirrorDbConfig.getMappingConfig().getGroupId())) { if (!mirrorDbConfig.getMappingConfig().getGroupId().equals(dml.getGroupId())) { continue; // 如果groupId不匹配则过滤 } 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 bf8d2de2..cc85b6b1 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 @@ -3,10 +3,7 @@ package com.alibaba.otter.canal.client.adapter.rdb.service; import java.sql.Connection; import java.sql.ResultSetMetaData; import java.sql.SQLException; -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.concurrent.*; import java.util.function.Function; @@ -138,7 +135,7 @@ public class RdbSyncService { * @param mappingConfig 配置集合 * @param dmls 批量 DML */ - public void sync(Map> mappingConfig, List dmls) { + public void sync(Map> mappingConfig, List dmls, Properties envProperties) { sync(dmls, dml -> { if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { // DDL @@ -150,28 +147,22 @@ public class RdbSyncService { String groupId = StringUtils.trimToEmpty(dml.getGroupId()); String database = dml.getDatabase(); String table = dml.getTable(); - Map configMap = mappingConfig - .get(destination + "-" + groupId + "_" + database + "-" + table); + Map configMap; + if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) { + configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table); + } else { + configMap = mappingConfig.get(destination + "_" + database + "-" + table); + } if (configMap == null) { return false; } - List configs = new ArrayList<>(); - configMap.values().forEach(config -> { - if (StringUtils.isNotEmpty(config.getGroupId())) { - if (config.getGroupId().equals(dml.getGroupId())) { - configs.add(config); - } - } else { - configs.add(config); - } - }); - if (configs.isEmpty()) { + if (configMap.values().isEmpty()) { return false; } - for (MappingConfig config : configs) { + for (MappingConfig config : configMap.values()) { if (config.getConcurrent()) { List singleDmls = SingleDml.dml2SingleDmls(dml); singleDmls.forEach(singleDml -> {