From b00d347eb077b0448f55981c9b76a1fe82f8a4ed Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 11 Jan 2019 13:40:37 +0800 Subject: [PATCH] =?UTF-8?q?rdb=20adapter=20=E5=A2=9E=E5=8A=A0=20truncate?= =?UTF-8?q?=20dml=E9=80=82=E9=85=8D=20fixed=20#1408=20fixed=20#1403?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/adapter/es/ESAdapter.java | 3 + .../client/adapter/hbase/HbaseAdapter.java | 3 + .../canal/client/adapter/rdb/RdbAdapter.java | 25 ++--- .../adapter/rdb/service/RdbSyncService.java | 96 +++++++++++-------- .../client/adapter/rdb/support/SingleDml.java | 37 +++++-- .../otter/canal/kafka/CanalKafkaProducer.java | 2 +- .../canal/rocketmq/CanalRocketMQProducer.java | 2 +- 7 files changed, 100 insertions(+), 68 deletions(-) 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 dd82cafc..a892cf04 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 @@ -127,6 +127,9 @@ public class ESAdapter implements OuterAdapter { } public void sync(List dmls) { + if (dmls == null || dmls.isEmpty()) { + return; + } for (Dml dml : dmls) { sync(dml); } 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 7201a2b9..95d00ca8 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 @@ -94,6 +94,9 @@ public class HbaseAdapter implements OuterAdapter { } public void sync(List dmls) { + if (dmls == null || dmls.isEmpty()) { + return; + } for (Dml dml : dmls) { sync(dml); } 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 33e659f0..2df46974 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 @@ -79,8 +79,8 @@ public class RdbAdapter implements OuterAdapter { // 过滤不匹配的key的配置 rdbMappingTmp.forEach((key, mappingConfig) -> { if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null) - || (mappingConfig.getOuterAdapterKey() != null && mappingConfig.getOuterAdapterKey() - .equalsIgnoreCase(configuration.getKey()))) { + || (mappingConfig.getOuterAdapterKey() != null + && mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) { rdbMapping.put(key, mappingConfig); } }); @@ -131,8 +131,8 @@ public class RdbAdapter implements OuterAdapter { String threads = properties.get("threads"); // String commitSize = properties.get("commitSize"); - boolean skipDupException = BooleanUtils.toBoolean(configuration.getProperties() - .getOrDefault("skipDupException", "true")); + boolean skipDupException = BooleanUtils + .toBoolean(configuration.getProperties().getOrDefault("skipDupException", "true")); rdbSyncService = new RdbSyncService(dataSource, threads != null ? Integer.valueOf(threads) : null, skipDupException); @@ -154,19 +154,12 @@ public class RdbAdapter implements OuterAdapter { */ @Override public void sync(List dmls) { - ExecutorService executorService = Executors.newFixedThreadPool(2); - - Future future1 = executorService.submit(() -> { - rdbSyncService.sync(mappingConfigCache, dmls); - return true; - }); - Future future2 = executorService.submit(() -> { - rdbMirrorDbSyncService.sync(dmls); - return true; - }); + if (dmls == null || dmls.isEmpty()) { + return; + } try { - future1.get(); - future2.get(); + rdbSyncService.sync(mappingConfigCache, dmls); + 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/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 2dde0ac2..612ba305 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 @@ -7,11 +7,7 @@ import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; +import java.util.concurrent.*; import java.util.function.Function; import javax.sql.DataSource; @@ -106,9 +102,8 @@ public class RdbSyncService { int j = i; 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; @@ -139,41 +134,41 @@ 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 database = dml.getDatabase(); - String table = dml.getTable(); - Map configMap = mappingConfig.get(destination + "." + database + "." + table); - - if (configMap == null) { + columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable()); return false; - } + } else { + // DML + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + Map configMap = mappingConfig.get(destination + "." + database + "." + table); - boolean executed = 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); - }); - } else { - int hash = 0; - List singleDmls = SingleDml.dml2SingleDmls(dml); - singleDmls.forEach(singleDml -> { - SyncItem syncItem = new SyncItem(config, singleDml); - dmlsPartition[hash].add(syncItem); - }); + if (configMap == null) { + return false; } - executed = true; + + boolean executed = 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); + }); + } else { + int hash = 0; + List singleDmls = SingleDml.dml2SingleDmls(dml); + singleDmls.forEach(singleDml -> { + SyncItem syncItem = new SyncItem(config, singleDml); + dmlsPartition[hash].add(syncItem); + }); + } + executed = true; + } + return executed; } - return executed; - } - } ); + }); } /** @@ -193,6 +188,8 @@ public class RdbSyncService { update(batchExecutor, config, dml); } else if (type != null && type.equalsIgnoreCase("DELETE")) { delete(batchExecutor, config, dml); + } else if (type != null && type.equalsIgnoreCase("TRUNCATE")) { + truncate(batchExecutor, config, dml); } if (logger.isDebugEnabled()) { logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); @@ -351,6 +348,23 @@ public class RdbSyncService { } } + /** + * truncate操作 + * + * @param config + */ + private void truncate(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException { + if (dml.getIsTruncate()) { + DbMapping dbMapping = config.getDbMapping(); + StringBuilder sql = new StringBuilder(); + sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(dbMapping)); + batchExecutor.execute(sql.toString(), new ArrayList<>()); + if (logger.isTraceEnabled()) { + logger.trace("Truncate target table, sql: {}", sql); + } + } + } + /** * 获取目标字段类型 * @@ -447,10 +461,10 @@ public class RdbSyncService { if (srcColumnName == null) { srcColumnName = Util.cleanColumn(targetColumnName); } - Object value; + Object value = null; if (o != null && o.containsKey(srcColumnName)) { value = o.get(srcColumnName); - } else { + } else if (d != null) { value = d.get(srcColumnName); } if (value != null) { 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 6e61a15f..417f2880 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 @@ -1,11 +1,11 @@ package com.alibaba.otter.canal.client.adapter.rdb.support; -import com.alibaba.otter.canal.client.adapter.support.Dml; - import java.util.ArrayList; import java.util.List; import java.util.Map; +import com.alibaba.otter.canal.client.adapter.support.Dml; + public class SingleDml { private String destination; @@ -14,6 +14,7 @@ public class SingleDml { private String type; private Map data; private Map old; + private boolean isTruncate = false; public String getDestination() { return destination; @@ -63,19 +64,37 @@ public class SingleDml { this.old = old; } + public boolean getIsTruncate() { + return isTruncate; + } + + public void setIsTruncate(boolean isTruncate) { + this.isTruncate = isTruncate; + } + public static List dml2SingleDmls(Dml dml) { - int size = dml.getData().size(); - List singleDmls = new ArrayList<>(size); - for (int i = 0; i < size; i++) { + int size = dml.getData() == null ? 0 : dml.getData().size(); + List singleDmls = new ArrayList<>(); + if (size > 0) { + for (int i = 0; i < size; i++) { + SingleDml singleDml = new SingleDml(); + singleDml.setDestination(dml.getDestination()); + singleDml.setDatabase(dml.getDatabase()); + singleDml.setTable(dml.getTable()); + singleDml.setType(dml.getType()); + singleDml.setData(dml.getData().get(i)); + if (dml.getOld() != null) { + singleDml.setOld(dml.getOld().get(i)); + } + singleDmls.add(singleDml); + } + } else if ("TRUNCATE".equalsIgnoreCase(dml.getType())) { SingleDml singleDml = new SingleDml(); singleDml.setDestination(dml.getDestination()); singleDml.setDatabase(dml.getDatabase()); singleDml.setTable(dml.getTable()); singleDml.setType(dml.getType()); - singleDml.setData(dml.getData().get(i)); - if (dml.getOld() != null) { - singleDml.setOld(dml.getOld().get(i)); - } + singleDml.setIsTruncate(true); singleDmls.add(singleDml); } return singleDmls; diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 64243dcb..611710d4 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -86,7 +86,7 @@ public class CanalKafkaProducer implements CanalMQProducer { .messageTopics(message, canalDestination.getTopic(), canalDestination.getDynamicTopic()); for (Map.Entry entry : messageMap.entrySet()) { - String topicName = entry.getKey(); + String topicName = entry.getKey().replace('.', '_'); Message messageSub = entry.getValue(); if (logger.isDebugEnabled()) { logger.debug("## Send message to kafka topic: " + topicName); diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index 953be44c..559704d1 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -67,7 +67,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { .messageTopics(data, destination.getTopic(), destination.getDynamicTopic()); for (Map.Entry entry : messageMap.entrySet()) { - String topicName = entry.getKey(); + String topicName = entry.getKey().replace('.', '_'); com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue(); send(destination, topicName, messageSub, callback); }