chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed #1433, optimzier code

Browse Source
This commit is contained in:
agapple
2019-03-17 10:35:16 +08:00
parent b114e0cb15
commit 95b26c4d1d
1 changed files with 45 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

@@ -107,8 +107,9 @@ 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;
@@ -146,49 +147,49 @@ 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;
columnsTypeCache.remove(dml.getDestination() + "." + dml.getDatabase() + "." + dml.getTable());
return false;
} else {
// DML
String destination = StringUtils.trimToEmpty(dml.getDestination());
String groupId = StringUtils.trimToEmpty(dml.getGroupId());
String database = dml.getDatabase();
String table = dml.getTable();
Map<String, MappingConfig> configMap;
if (envProperties != null && !"tcp".equalsIgnoreCase(envProperties.getProperty("canal.conf.mode"))) {
configMap = mappingConfig.get(destination + "-" + groupId + "_" + database + "-" + table);
} else {
// DML
String destination = StringUtils.trimToEmpty(dml.getDestination());
String groupId = StringUtils.trimToEmpty(dml.getGroupId());
String database = dml.getDatabase();
String table = dml.getTable();
Map<String, MappingConfig> 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;
}
if (configMap.values().isEmpty()) {
return 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);
});
}
}
return true;
configMap = mappingConfig.get(destination + "_" + database + "-" + table);
}
});
if (configMap == null) {
return false;
}
if (configMap.values().isEmpty()) {
return 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);
});
}
}
return true;
}
} );
}
/**
@@ -314,7 +315,7 @@ public class RdbSyncService {
for (String srcColumnName : old.keySet()) {
List<String> targetColumnNames = new ArrayList<>();
columnsMap.forEach((targetColumn, srcColumn) -> {
if (srcColumnName.toLowerCase().equals(srcColumn.toLowerCase())) {
if (srcColumnName.equalsIgnoreCase(srcColumn)) {
targetColumnNames.add(targetColumn);
}
});