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

Merge branch 'master' into master

Browse Source
This commit is contained in:
rewerma
2019-03-21 19:42:07 +08:00
committed by GitHub
parent 4bad0e88f8 024ffc2aa2
commit 38d8c3d462
38 changed files with 625 additions and 2957 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java
View File

@@ -2,7 +2,10 @@ package com.alibaba.otter.canal.client.adapter.rdb;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import javax.sql.DataSource;
@@ -11,6 +14,7 @@ import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
@@ -22,8 +26,12 @@ import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService;
import com.alibaba.otter.canal.client.adapter.rdb.service.RdbMirrorDbSyncService;
import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService;
import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
import com.alibaba.otter.canal.client.adapter.support.*;
import org.slf4j.MDC;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.client.adapter.support.Util;
/**
* RDB适配器实现类
@@ -74,8 +82,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);
}
});
@@ -122,9 +130,9 @@ public class RdbAdapter implements OuterAdapter {
dataSource.setTimeBetweenEvictionRunsMillis(60000);
dataSource.setMinEvictableIdleTimeMillis(300000);
dataSource.setUseUnfairLock(true);
List<String> array = new ArrayList<>();
array.add("set names utf8mb4;");
dataSource.setConnectionInitSqls(array);
// List<String> array = new ArrayList<>();
// array.add("set names utf8mb4;");
// dataSource.setConnectionInitSqls(array);
try {
dataSource.init();
@@ -135,8 +143,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);

20
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java
View File

@@ -52,17 +52,17 @@ public class BatchExecutor implements Closeable {
}
public void execute(String sql, List<Map<String, ?>> values) throws SQLException {
try (PreparedStatement pstmt = getConn().prepareStatement(sql)) {
int len = values.size();
for (int i = 0; i < len; i++) {
int type = (Integer) values.get(i).get("type");
Object value = values.get(i).get("value");
SyncUtil.setPStmt(type, pstmt, value, i + 1);
}
pstmt.execute();
idx.incrementAndGet();
PreparedStatement pstmt = getConn().prepareStatement(sql);
int len = values.size();
for (int i = 0; i < len; i++) {
int type = (Integer) values.get(i).get("type");
Object value = values.get(i).get("value");
SyncUtil.setPStmt(type, pstmt, value, i + 1);
}
pstmt.execute();
idx.incrementAndGet();
pstmt.close();
}
public void commit() throws SQLException {