diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java index 38fd7a3b..97204867 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java @@ -15,6 +15,7 @@ public class Dml implements Serializable { private static final long serialVersionUID = 2611556444074013268L; private String destination; // 对应canal的实例或者MQ的topic + private String groupId; // 对应mq的group id private String database; // 数据库或schema private String table; // 表名 private List pkNames; @@ -36,6 +37,14 @@ public class Dml implements Serializable { this.destination = destination; } + public String getGroupId() { + return groupId; + } + + public void setGroupId(String groupId) { + this.groupId = groupId; + } + public String getDatabase() { return database; } 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 1ddbae0f..239b8839 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 @@ -14,7 +14,7 @@ import com.alibaba.otter.canal.protocol.Message; */ public class MessageUtil { - public static List parse4Dml(String destination, Message message) { + public static List parse4Dml(String destination, String groupId, Message message) { if (message == null) { return null; } @@ -38,6 +38,7 @@ public class MessageUtil { final Dml dml = new Dml(); dml.setDestination(destination); + dml.setGroupId(groupId); dml.setDatabase(entry.getHeader().getSchemaName()); dml.setTable(entry.getHeader().getTableName()); dml.setType(eventType.toString()); @@ -118,10 +119,10 @@ public class MessageUtil { return dmls; } - public static List flatMessage2Dml(String destination, List flatMessages) { + public static List flatMessage2Dml(String destination, String groupId, List flatMessages) { List dmls = new ArrayList(flatMessages.size()); for (FlatMessage flatMessage : flatMessages) { - Dml dml = flatMessage2Dml(destination, flatMessage); + Dml dml = flatMessage2Dml(destination, groupId, flatMessage); if (dml != null) { dmls.add(dml); } @@ -130,12 +131,13 @@ public class MessageUtil { return dmls; } - public static Dml flatMessage2Dml(String destination, FlatMessage flatMessage) { + public static Dml flatMessage2Dml(String destination, String groupId, FlatMessage flatMessage) { if (flatMessage == null) { return null; } Dml dml = new Dml(); dml.setDestination(destination); + dml.setGroupId(groupId); dml.setDatabase(flatMessage.getDatabase()); dml.setTable(flatMessage.getTable()); dml.setPkNames(flatMessage.getPkNames()); 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 b22ecdca..ab57c1d8 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 @@ -8,6 +8,7 @@ import java.util.regex.Pattern; import javax.sql.DataSource; +import org.apache.commons.lang.StringUtils; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.common.settings.Settings; @@ -133,7 +134,19 @@ public class ESAdapter implements OuterAdapter { String table = dml.getTable(); Map configMap = dbTableEsSyncConfig.get(database + "-" + table); if (configMap != null) { - esSyncService.sync(configMap.values(), dml); + List configs = new ArrayList<>(); + configMap.values().forEach(esConfig -> { + if (StringUtils.isNotEmpty(esConfig.getGroupId())) { + if (esConfig.getGroupId().equals(dml.getGroupId())) { + configs.add(esConfig); + } + } else { + configs.add(esConfig); + } + }); + if (!configs.isEmpty()) { + esSyncService.sync(configs, dml); + } } } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java index 65087437..09c2f4e0 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfig.java @@ -17,6 +17,8 @@ public class ESSyncConfig { private String outerAdapterKey; // adapter key + private String groupId; // group id + private String destination; // canal destination private ESMapping esMapping; @@ -52,6 +54,14 @@ public class ESSyncConfig { this.outerAdapterKey = outerAdapterKey; } + public String getGroupId() { + return groupId; + } + + public void setGroupId(String groupId) { + this.groupId = groupId; + } + public String getDestination() { return destination; } 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 45b1bd51..b54f8f46 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 @@ -108,7 +108,19 @@ public class HbaseAdapter implements OuterAdapter { String table = dml.getTable(); Map configMap = mappingConfigCache.get(destination + "." + database + "." + table); if (configMap != null) { - configMap.values().forEach(config -> hbaseSyncService.sync(config, dml)); + 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()) { + configs.forEach(config -> hbaseSyncService.sync(config, dml)); + } } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java index 2d86adee..3d1c38be 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java @@ -14,6 +14,8 @@ public class MappingConfig { private String outerAdapterKey; // adapter key + private String groupId; // groupId + private String destination; // canal实例或MQ的topic private HbaseMapping hbaseMapping; // hbase映射配置 @@ -26,6 +28,14 @@ public class MappingConfig { this.dataSourceKey = dataSourceKey; } + public String getGroupId() { + return groupId; + } + + public void setGroupId(String groupId) { + this.groupId = groupId; + } + public String getOuterAdapterKey() { return outerAdapterKey; } 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 e4d58272..6f21ebd5 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 @@ -31,6 +31,7 @@ public abstract class AbstractCanalAdapterWorker { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); protected String canalDestination; // canal实例 + protected String groupId = null; // groupId protected List> canalOuterAdapters; // 外部适配器 protected CanalClientConfig canalClientConfig; // 配置 protected ExecutorService groupInnerExecutorService; // 组内工作线程池 @@ -56,7 +57,7 @@ public abstract class AbstractCanalAdapterWorker { // 组内适配器穿行运行，尽量不要配置组内适配器 adapters.forEach(adapter -> { long begin = System.currentTimeMillis(); - List dmls = MessageUtil.parse4Dml(canalDestination, message); + List dmls = MessageUtil.parse4Dml(canalDestination, groupId, message); if (dmls != null) { batchSync(dmls, adapter); @@ -101,7 +102,7 @@ public abstract class AbstractCanalAdapterWorker { // 组内适配器穿行运行，尽量不要配置组内适配器 outerAdapters.forEach(adapter -> { long begin = System.currentTimeMillis(); - List dmls = MessageUtil.flatMessage2Dml(canalDestination, flatMessages); + List dmls = MessageUtil.flatMessage2Dml(canalDestination, groupId, flatMessages); batchSync(dmls, adapter); if (logger.isDebugEnabled()) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index 58ffed8f..72291ccb 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -28,7 +28,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { super(canalOuterAdapters); this.canalClientConfig = canalClientConfig; this.topic = topic; - this.canalDestination = topic; + super.canalDestination = topic; + super.groupId = groupId; this.flatMessage = flatMessage; this.connector = new KafkaCanalConnector(bootstrapServers, topic, @@ -45,10 +46,12 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { try { Thread.sleep(1000); } catch (InterruptedException e) { + // ignore } } ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); - int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); + int retry = canalClientConfig.getRetries() == null + || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 while (running) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java index b6f276b2..369e8965 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -28,7 +28,8 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { this.canalClientConfig = canalClientConfig; this.topic = topic; this.flatMessage = flatMessage; - this.canalDestination = topic; + super.canalDestination = topic; + super.groupId = groupId; this.connector = new RocketMQCanalConnector(nameServers, topic, groupId, 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 5c0e9f9f..d2421020 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 @@ -17,6 +17,8 @@ public class MappingConfig { private String destination; // canal实例或MQ的topic + private String groupId; // groupId + private String outerAdapterKey; // 对应适配器的key private Boolean concurrent; // 是否并行同步 @@ -31,6 +33,14 @@ public class MappingConfig { this.dataSourceKey = dataSourceKey; } + public String getGroupId() { + return groupId; + } + + public void setGroupId(String groupId) { + this.groupId = groupId; + } + public String getOuterAdapterKey() { return outerAdapterKey; } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java index d84f66e7..dbead72e 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MirrorDbConfig.java @@ -7,7 +7,7 @@ public class MirrorDbConfig { private String fileName; private MappingConfig mappingConfig; - Map tableConfig = new ConcurrentHashMap<>(); + private Map tableConfig = new ConcurrentHashMap<>(); public static MirrorDbConfig create(String fileName, MappingConfig mappingConfig) { return new MirrorDbConfig(fileName, mappingConfig); 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 16d8af5f..eec96e0c 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 @@ -56,6 +56,15 @@ public class RdbMirrorDbSyncService { if (mirrorDbConfig == null) { continue; } + if (mirrorDbConfig.getMappingConfig() == null) { + continue; + } + if (StringUtils.isNotEmpty(mirrorDbConfig.getMappingConfig().getGroupId())) { + if (!mirrorDbConfig.getMappingConfig().getGroupId().equals(dml.getGroupId())) { + continue; // 如果groupId不匹配则过滤 + } + } + if (dml.getIsDdl() != null && dml.getIsDdl() && StringUtils.isNotEmpty(dml.getSql())) { // DDL if (logger.isDebugEnabled()) { @@ -71,37 +80,35 @@ public class RdbMirrorDbSyncService { } } if (!dmlList.isEmpty()) { - rdbSyncService.sync(dmlList, - dml -> { - MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(dml.getDestination() + "." - + dml.getDatabase()); - if (mirrorDbConfig == null) { - return false; - } - String table = dml.getTable(); - MappingConfig config = mirrorDbConfig.getTableConfig().get(table); + rdbSyncService.sync(dmlList, dml -> { + MirrorDbConfig mirrorDbConfig = mirrorDbConfigCache.get(dml.getDestination() + "." + dml.getDatabase()); + if (mirrorDbConfig == null) { + return false; + } + String table = dml.getTable(); + MappingConfig config = mirrorDbConfig.getTableConfig().get(table); - if (config == null) { - return false; - } + if (config == null) { + return false; + } - if (config.getConcurrent()) { - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData()); - RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); - rdbSyncService.getDmlsPartition()[hash].add(syncItem); - }); - } else { - int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); - rdbSyncService.getDmlsPartition()[hash].add(syncItem); - }); - } - return true; - }); + if (config.getConcurrent()) { + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + int hash = rdbSyncService.pkHash(config.getDbMapping(), singleDml.getData()); + RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); + rdbSyncService.getDmlsPartition()[hash].add(syncItem); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + RdbSyncService.SyncItem syncItem = new RdbSyncService.SyncItem(config, singleDml); + rdbSyncService.getDmlsPartition()[hash].add(syncItem); + }); + } + return true; + }); } } @@ -119,6 +126,7 @@ public class RdbMirrorDbSyncService { mappingConfig = new MappingConfig(); mappingConfig.setDataSourceKey(baseConfigMap.getDataSourceKey()); mappingConfig.setDestination(baseConfigMap.getDestination()); + mappingConfig.setGroupId(baseConfigMap.getGroupId()); mappingConfig.setOuterAdapterKey(baseConfigMap.getOuterAdapterKey()); mappingConfig.setConcurrent(baseConfigMap.getConcurrent()); MappingConfig.DbMapping dbMapping = new MappingConfig.DbMapping(); 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 d2ba7dac..68e79c8e 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 @@ -155,8 +155,22 @@ public class RdbSyncService { 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()) { + return false; + } + boolean executed = false; - for (MappingConfig config : configMap.values()) { + for (MappingConfig config : configs) { if (config.getConcurrent()) { List singleDmls = SingleDml.dml2SingleDmls(dml); singleDmls.forEach(singleDml -> {