mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
@@ -1,5 +1,7 @@
|
|||||||
package com.alibaba.otter.canal.client.adapter.rdb.config;
|
package com.alibaba.otter.canal.client.adapter.rdb.config;
|
||||||
|
|
||||||
|
import org.apache.commons.lang.StringUtils;
|
||||||
|
|
||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
|
||||||
@@ -92,7 +94,7 @@ public class MappingConfig {
|
|||||||
private Map<String, String> allMapColumns;
|
private Map<String, String> allMapColumns;
|
||||||
|
|
||||||
public Boolean getMirrorDb() {
|
public Boolean getMirrorDb() {
|
||||||
return mirrorDb;
|
return mirrorDb == null ? false : mirrorDb;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setMirrorDb(Boolean mirrorDb) {
|
public void setMirrorDb(Boolean mirrorDb) {
|
||||||
@@ -148,6 +150,13 @@ public class MappingConfig {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public Map<String, String> getTargetColumns() {
|
public Map<String, String> getTargetColumns() {
|
||||||
|
if (targetColumns != null) {
|
||||||
|
targetColumns.forEach((key, value) -> {
|
||||||
|
if (StringUtils.isEmpty(value)) {
|
||||||
|
targetColumns.put(key, key);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
return targetColumns;
|
return targetColumns;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import java.io.File;
|
|||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
|
||||||
import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig;
|
|
||||||
import org.apache.commons.io.filefilter.FileFilterUtils;
|
import org.apache.commons.io.filefilter.FileFilterUtils;
|
||||||
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
|
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
|
||||||
import org.apache.commons.io.monitor.FileAlterationMonitor;
|
import org.apache.commons.io.monitor.FileAlterationMonitor;
|
||||||
@@ -16,6 +15,7 @@ import org.yaml.snakeyaml.Yaml;
|
|||||||
|
|
||||||
import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
|
import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
|
||||||
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
|
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
|
||||||
|
import com.alibaba.otter.canal.client.adapter.rdb.config.MirrorDbConfig;
|
||||||
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
|
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
|
||||||
import com.alibaba.otter.canal.client.adapter.support.Util;
|
import com.alibaba.otter.canal.client.adapter.support.Util;
|
||||||
|
|
||||||
@@ -121,6 +121,9 @@ public class RdbConfigMonitor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void addConfigToCache(File file, MappingConfig mappingConfig) {
|
private void addConfigToCache(File file, MappingConfig mappingConfig) {
|
||||||
|
MappingConfig.DbMapping dbMapping = mappingConfig.getDbMapping();
|
||||||
|
if (dbMapping != null) {
|
||||||
|
if (!dbMapping.getMirrorDb()) {
|
||||||
rdbAdapter.getRdbMapping().put(file.getName(), mappingConfig);
|
rdbAdapter.getRdbMapping().put(file.getName(), mappingConfig);
|
||||||
Map<String, MappingConfig> configMap = rdbAdapter.getMappingConfigCache()
|
Map<String, MappingConfig> configMap = rdbAdapter.getMappingConfigCache()
|
||||||
.computeIfAbsent(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
|
.computeIfAbsent(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
|
||||||
@@ -128,12 +131,14 @@ public class RdbConfigMonitor {
|
|||||||
+ mappingConfig.getDbMapping().getTable(),
|
+ mappingConfig.getDbMapping().getTable(),
|
||||||
k1 -> new HashMap<>());
|
k1 -> new HashMap<>());
|
||||||
configMap.put(file.getName(), mappingConfig);
|
configMap.put(file.getName(), mappingConfig);
|
||||||
|
} else {
|
||||||
Map<String, MirrorDbConfig> mirrorDbConfigCache = rdbAdapter.getMirrorDbConfigCache();
|
Map<String, MirrorDbConfig> mirrorDbConfigCache = rdbAdapter.getMirrorDbConfigCache();
|
||||||
mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
|
mirrorDbConfigCache.put(StringUtils.trimToEmpty(mappingConfig.getDestination()) + "."
|
||||||
+ mappingConfig.getDbMapping().getDatabase(),
|
+ mappingConfig.getDbMapping().getDatabase(),
|
||||||
MirrorDbConfig.create(file.getName(), mappingConfig));
|
MirrorDbConfig.create(file.getName(), mappingConfig));
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private void deleteConfigFromCache(File file) {
|
private void deleteConfigFromCache(File file) {
|
||||||
|
|
||||||
|
|||||||
@@ -193,6 +193,7 @@
|
|||||||
|
|
||||||
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
||||||
<property name="topic" value="${canal.mq.topic}" />
|
<property name="topic" value="${canal.mq.topic}" />
|
||||||
|
<property name="dynamicTopic" value="${canal.mq.dynamicTopic}" />
|
||||||
<property name="partition" value="${canal.mq.partition}" />
|
<property name="partition" value="${canal.mq.partition}" />
|
||||||
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
||||||
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
||||||
|
|||||||
@@ -178,6 +178,7 @@
|
|||||||
|
|
||||||
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
||||||
<property name="topic" value="${canal.mq.topic}" />
|
<property name="topic" value="${canal.mq.topic}" />
|
||||||
|
<property name="dynamicTopic" value="${canal.mq.dynamicTopic}" />
|
||||||
<property name="partition" value="${canal.mq.partition}" />
|
<property name="partition" value="${canal.mq.partition}" />
|
||||||
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
||||||
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
||||||
|
|||||||
@@ -268,6 +268,7 @@
|
|||||||
|
|
||||||
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
||||||
<property name="topic" value="${canal.mq.topic}" />
|
<property name="topic" value="${canal.mq.topic}" />
|
||||||
|
<property name="dynamicTopic" value="${canal.mq.dynamicTopic}" />
|
||||||
<property name="partition" value="${canal.mq.partition}" />
|
<property name="partition" value="${canal.mq.partition}" />
|
||||||
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
||||||
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
||||||
|
|||||||
@@ -166,6 +166,7 @@
|
|||||||
|
|
||||||
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
<bean id="mqConfig" class="com.alibaba.otter.canal.instance.core.CanalMQConfig">
|
||||||
<property name="topic" value="${canal.mq.topic}" />
|
<property name="topic" value="${canal.mq.topic}" />
|
||||||
|
<property name="dynamicTopic" value="${canal.mq.dynamicTopic}" />
|
||||||
<property name="partition" value="${canal.mq.partition}" />
|
<property name="partition" value="${canal.mq.partition}" />
|
||||||
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
<property name="partitionsNum" value="${canal.mq.partitionsNum}" />
|
||||||
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
<property name="partitionHash" value="${canal.mq.partitionHash}" />
|
||||||
|
|||||||
Reference in New Issue
Block a user