Merge pull request #1412 from rewerma/master

fixed #1408 fixed # 1403
2026-11-04 00:30:15 +08:00 · 2019-01-12 11:26:11 +08:00
parent f1753e79f9 b00d347eb0
commit 70a4f0363d
7 changed files with 100 additions and 68 deletions
--- 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<Dml> dmls) {
+        if (dmls == null || dmls.isEmpty()) {
+            return;
+        }
        for (Dml dml : dmls) {
            sync(dml);
        }
--- 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<Dml> dmls) {
+        if (dmls == null || dmls.isEmpty()) {
+            return;
+        }
        for (Dml dml : dmls) {
            sync(dml);
        }
--- 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<Dml> dmls) {
-        ExecutorService executorService = Executors.newFixedThreadPool(2);
-
-        Future<Boolean> future1 = executorService.submit(() -> {
-            rdbSyncService.sync(mappingConfigCache, dmls);
-            return true;
-        });
-        Future<Boolean> 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);
        }
--- 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<String, MappingConfig> 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<String, MappingConfig> configMap = mappingConfig.get(destination + "." + database + "." + table);

-            boolean executed = false;
-            for (MappingConfig config : configMap.values()) {
-                if (config.getConcurrent()) {
-                    List<SingleDml> 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<SingleDml> 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<SingleDml> 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<SingleDml> 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) {
--- 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<String, Object> data;
    private Map<String, Object> 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<SingleDml> dml2SingleDmls(Dml dml) {
-        int size = dml.getData().size();
-        List<SingleDml> singleDmls = new ArrayList<>(size);
-        for (int i = 0; i < size; i++) {
+        int size = dml.getData() == null ? 0 : dml.getData().size();
+        List<SingleDml> 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;
--- 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<String, Message> 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);
--- 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<String, com.alibaba.otter.canal.protocol.Message> 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);
            }