format clickhouse adapter code

This commit is contained in:
jianghang.loujh
2023-12-04 09:46:23 +08:00
parent 9baeff55e4
commit 93dac2d335
15 changed files with 206 additions and 151 deletions

View File

@@ -1,5 +1,15 @@
package com.alibaba.otter.canal.client.adapter.clickhouse; package com.alibaba.otter.canal.client.adapter.clickhouse;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.filter.stat.StatFilter; import com.alibaba.druid.filter.stat.StatFilter;
import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.util.JdbcUtils; import com.alibaba.druid.util.JdbcUtils;
@@ -13,15 +23,6 @@ import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseEtlSe
import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseMirrorDbBatchSyncService; import com.alibaba.otter.canal.client.adapter.clickhouse.service.ClickHouseMirrorDbBatchSyncService;
import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
import com.alibaba.otter.canal.client.adapter.support.*; import com.alibaba.otter.canal.client.adapter.support.*;
import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
/** /**
* ClickHouse Adapter implementation * ClickHouse Adapter implementation
@@ -35,11 +36,15 @@ public class ClickHouseAdapter implements OuterAdapter {
private static final Logger logger = LoggerFactory.getLogger(ClickHouseAdapter.class); private static final Logger logger = LoggerFactory.getLogger(ClickHouseAdapter.class);
private Map<String, MappingConfig> clickHouseMapping = new ConcurrentHashMap<>(); // Store the mapping of filename and configuration, load yml files below resource path // Store the mapping of filename and configuration, load yml files below
// resource path
private Map<String, MappingConfig> clickHouseMapping = new ConcurrentHashMap<>();
private Map<String, Map<String, MappingConfig>> mappingConfigCache = new ConcurrentHashMap<>(); // Schema -> Table -> MappingConfig // Schema -> Table -> MappingConfig
private Map<String, Map<String, MappingConfig>> mappingConfigCache = new ConcurrentHashMap<>();
private Map<String, MirrorDbConfig> mirrorDbConfigCache = new ConcurrentHashMap<>(); // Mirror DB Configuration, don't need to load column mapping // Mirror DB Configuration, don't need to load column mapping
private Map<String, MirrorDbConfig> mirrorDbConfigCache = new ConcurrentHashMap<>();
private DruidDataSource dataSource; private DruidDataSource dataSource;
@@ -49,7 +54,8 @@ public class ClickHouseAdapter implements OuterAdapter {
private Properties envProperties; private Properties envProperties;
private OuterAdapterConfig configuration; // Launch configuration // Launch configuration
private OuterAdapterConfig configuration;
private ClickHouseConfigMonitor clickHouseConfigMonitor; private ClickHouseConfigMonitor clickHouseConfigMonitor;

View File

@@ -1,19 +1,22 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.config; package com.alibaba.otter.canal.client.adapter.clickhouse.config;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import java.util.Properties; import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
/** /**
* CLICKHOUSE表映射配置加载器 * CLICKHOUSE表映射配置加载器
* *
* @author rewerma 2018-11-07 下午02:41:34 * @author: Xander
* @version 1.0.0 * @date: Created in 2023/11/10 22:23
* @email: zhrunxin33@gmail.com
* @version 1.1.8
*/ */
public class ConfigLoader { public class ConfigLoader {

View File

@@ -1,16 +1,19 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.config; package com.alibaba.otter.canal.client.adapter.clickhouse.config;
import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
import org.apache.commons.lang.StringUtils;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
/** /**
* CLICKHOUSE表映射配置 * CLICKHOUSE表映射配置
* *
* @author rewerma 2018-11-07 下午02:41:34 * @author: Xander
* @version 1.0.0 * @date: Created in 2023/11/10 22:23
* @email: zhrunxin33@gmail.com
* @version 1.1.8
*/ */
public class MappingConfig implements AdapterConfig { public class MappingConfig implements AdapterConfig {

View File

@@ -3,10 +3,17 @@ package com.alibaba.otter.canal.client.adapter.clickhouse.config;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
/**
* @author: Xander
* @date: Created in 2023/11/10 22:23
* @email: zhrunxin33@gmail.com
* @version 1.1.8
*/
public class MirrorDbConfig { public class MirrorDbConfig {
private String fileName; private String fileName;
private MappingConfig mappingConfig; private MappingConfig mappingConfig;
private Map<String, MappingConfig> tableConfig = new ConcurrentHashMap<>(); private Map<String, MappingConfig> tableConfig = new ConcurrentHashMap<>();
public static MirrorDbConfig create(String fileName, MappingConfig mappingConfig) { public static MirrorDbConfig create(String fileName, MappingConfig mappingConfig) {

View File

@@ -1,10 +1,8 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.monitor; package com.alibaba.otter.canal.client.adapter.clickhouse.monitor;
import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter; import java.io.File;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig; import java.util.Properties;
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.YamlUtils;
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;
@@ -12,8 +10,18 @@ import org.apache.commons.io.monitor.FileAlterationObserver;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.io.File; import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter;
import java.util.Properties; import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
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.YamlUtils;
/**
* @author: Xander
* @date: Created in 2023/11/10 22:23
* @email: zhrunxin33@gmail.com
* @version 1.1.8
*/
public class ClickHouseConfigMonitor { public class ClickHouseConfigMonitor {

View File

@@ -1,5 +1,18 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.service; package com.alibaba.otter.canal.client.adapter.clickhouse.service;
import java.sql.Connection;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Types;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature; import com.alibaba.fastjson2.JSONWriter.Feature;
@@ -10,18 +23,6 @@ import com.alibaba.otter.canal.client.adapter.clickhouse.support.SingleDml;
import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.Util; import com.alibaba.otter.canal.client.adapter.support.Util;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Types;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;
/** /**
* ClickHouse batch synchronize * ClickHouse batch synchronize
@@ -29,6 +30,7 @@ import java.util.stream.Collectors;
* @author: Xander * @author: Xander
* @date: Created in 2023/11/10 22:23 * @date: Created in 2023/11/10 22:23
* @email: zhrunxin33@gmail.com * @email: zhrunxin33@gmail.com
* @version 1.1.8
*/ */
public class ClickHouseBatchSyncService { public class ClickHouseBatchSyncService {
@@ -45,7 +47,7 @@ public class ClickHouseBatchSyncService {
private BatchExecutor alterExecutors; // Alter Single Executor(update/delete/truncate) private BatchExecutor alterExecutors; // Alter Single Executor(update/delete/truncate)
private ExecutorService[] executorThreads; // Be initialized once private ExecutorService[] executorThreads; // Be initialized once
private ScheduledExecutorService[] scheduledExecutors; private ScheduledExecutorService[] scheduledExecutors;
private int threads = 3; // Default parallel thread count private int threads = 3; // Default parallel thread count

View File

@@ -1,5 +1,14 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.service; package com.alibaba.otter.canal.client.adapter.clickhouse.service;
import java.sql.*;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import javax.sql.DataSource;
import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig.DbMapping; import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig.DbMapping;
@@ -9,19 +18,13 @@ import com.alibaba.otter.canal.client.adapter.support.AdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.Util; import com.alibaba.otter.canal.client.adapter.support.Util;
import javax.sql.DataSource;
import java.sql.*;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
/** /**
* ClickHouse ETL 操作业务类 * ClickHouse ETL 操作业务类
* *
* @author rewerma @ 2018-11-7 * @author: Xander
* @version 1.0.0 * @date: Created in 2023/11/10 22:23
* @email: zhrunxin33@gmail.com
* @version 1.1.8
*/ */
public class ClickHouseEtlService extends AbstractEtlService { public class ClickHouseEtlService extends AbstractEtlService {
@@ -58,23 +61,23 @@ public class ClickHouseEtlService extends AbstractEtlService {
Util.sqlRS(targetDS, Util.sqlRS(targetDS,
"SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " LIMIT 1 ", "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " LIMIT 1 ",
rs -> { rs -> {
try { try {
ResultSetMetaData rsd = rs.getMetaData(); ResultSetMetaData rsd = rs.getMetaData();
int columnCount = rsd.getColumnCount(); int columnCount = rsd.getColumnCount();
List<String> columns = new ArrayList<>(); List<String> columns = new ArrayList<>();
for (int i = 1; i <= columnCount; i++) { for (int i = 1; i <= columnCount; i++) {
columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i)); columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i));
columns.add(rsd.getColumnName(i)); columns.add(rsd.getColumnName(i));
}
columnsMap.putAll(SyncUtil.getColumnsMap(dbMapping, columns));
return true;
} catch (Exception e) {
logger.error(e.getMessage(), e);
return false;
} }
});
columnsMap.putAll(SyncUtil.getColumnsMap(dbMapping, columns));
return true;
} catch (Exception e) {
logger.error(e.getMessage(), e);
return false;
}
});
Util.sqlRS(srcDS, sql, values, rs -> { Util.sqlRS(srcDS, sql, values, rs -> {
int idx = 1; int idx = 1;
@@ -86,8 +89,10 @@ public class ClickHouseEtlService extends AbstractEtlService {
insertSql.append("INSERT INTO ") insertSql.append("INSERT INTO ")
.append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())) .append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()))
.append(" ("); .append(" (");
columnsMap columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick)
.forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick).append(targetColumnName).append(backtick).append(",")); .append(targetColumnName)
.append(backtick)
.append(","));
int len = insertSql.length(); int len = insertSql.length();
insertSql.delete(len - 1, len).append(") VALUES ("); insertSql.delete(len - 1, len).append(") VALUES (");
@@ -97,7 +102,7 @@ public class ClickHouseEtlService extends AbstractEtlService {
} }
len = insertSql.length(); len = insertSql.length();
insertSql.delete(len - 1, len).append(")"); insertSql.delete(len - 1, len).append(")");
logger.info("executeSqlImport sql:{}",insertSql.toString()); logger.info("executeSqlImport sql:{}", insertSql.toString());
try (Connection connTarget = targetDS.getConnection(); try (Connection connTarget = targetDS.getConnection();
PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) { PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
connTarget.setAutoCommit(false); connTarget.setAutoCommit(false);
@@ -110,7 +115,8 @@ public class ClickHouseEtlService extends AbstractEtlService {
// 删除数据 // 删除数据
Map<String, Object> pkVal = new LinkedHashMap<>(); Map<String, Object> pkVal = new LinkedHashMap<>();
StringBuilder deleteSql = new StringBuilder( StringBuilder deleteSql = new StringBuilder(
"ALTER TABLE " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " DELETE WHERE "); "ALTER TABLE " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())
+ " DELETE WHERE ");
appendCondition(dbMapping, deleteSql, pkVal, rs, backtick); appendCondition(dbMapping, deleteSql, pkVal, rs, backtick);
try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
int k = 1; int k = 1;

View File

@@ -1,16 +1,5 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.service; package com.alibaba.otter.canal.client.adapter.clickhouse.service;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MirrorDbConfig;
import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection; import java.sql.Connection;
import java.sql.Statement; import java.sql.Statement;
import java.util.ArrayList; import java.util.ArrayList;
@@ -18,26 +7,48 @@ import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MirrorDbConfig;
import com.alibaba.otter.canal.client.adapter.clickhouse.support.SyncUtil;
import com.alibaba.otter.canal.client.adapter.support.Dml;
/** /**
* CLICKHOUSE镜像库同步操作业务 * CLICKHOUSE镜像库同步操作业务
* *
* @author rewerma 2018-12-12 下午011:23 * @author: Xander
* @version 1.0.0 * @date: Created in 2023/11/10 22:23
* @email: zhrunxin33@gmail.com
* @version 1.1.8
*/ */
public class ClickHouseMirrorDbBatchSyncService { public class ClickHouseMirrorDbBatchSyncService {
private static final Logger logger = LoggerFactory.getLogger(ClickHouseMirrorDbBatchSyncService.class); private static final Logger logger = LoggerFactory.getLogger(ClickHouseMirrorDbBatchSyncService.class);
private Map<String, MirrorDbConfig> mirrorDbConfigCache; // 镜像库配置 // 镜像库配置
private Map<String, MirrorDbConfig> mirrorDbConfigCache;
private DruidDataSource dataSource; private DruidDataSource dataSource;
private ClickHouseBatchSyncService clickHouseBatchSyncService; // clickhouseSyncService代理 // clickhouseSyncService代理
private ClickHouseBatchSyncService clickHouseBatchSyncService;
public ClickHouseMirrorDbBatchSyncService(Map<String, MirrorDbConfig> mirrorDbConfigCache, DruidDataSource dataSource, public ClickHouseMirrorDbBatchSyncService(Map<String, MirrorDbConfig> mirrorDbConfigCache,
Integer threads, Integer batchSize, Long scheduleTime, Map<String, Map<String, Integer>> columnsTypeCache, DruidDataSource dataSource, Integer threads, Integer batchSize,
Long scheduleTime, Map<String, Map<String, Integer>> columnsTypeCache,
boolean skipDupException){ boolean skipDupException){
this.mirrorDbConfigCache = mirrorDbConfigCache; this.mirrorDbConfigCache = mirrorDbConfigCache;
this.dataSource = dataSource; this.dataSource = dataSource;
this.clickHouseBatchSyncService = new ClickHouseBatchSyncService(dataSource, threads, batchSize, scheduleTime, columnsTypeCache, skipDupException); this.clickHouseBatchSyncService = new ClickHouseBatchSyncService(dataSource,
threads,
batchSize,
scheduleTime,
columnsTypeCache,
skipDupException);
} }
/** /**
@@ -73,7 +84,8 @@ public class ClickHouseMirrorDbBatchSyncService {
logger.debug("DDL: {}", JSON.toJSONString(dml, Feature.WriteNulls)); logger.debug("DDL: {}", JSON.toJSONString(dml, Feature.WriteNulls));
} }
executeDdl(mirrorDbConfig, dml); executeDdl(mirrorDbConfig, dml);
clickHouseBatchSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable()); clickHouseBatchSyncService.getColumnsTypeCache()
.remove(destination + "." + database + "." + dml.getTable());
mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置 mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置
} else { } else {
// DML // DML

View File

@@ -1,9 +1,5 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.support; package com.alibaba.otter.canal.client.adapter.clickhouse.support;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.sql.DataSource;
import java.io.Closeable; import java.io.Closeable;
import java.sql.Connection; import java.sql.Connection;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
@@ -13,6 +9,11 @@ import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/** /**
* sql批量执行器 * sql批量执行器
* *
@@ -64,7 +65,7 @@ public class BatchExecutor implements Closeable {
pstmt.close(); pstmt.close();
} }
public void batchExecute(String sql, List<List<Map<String,?>>> batchValues) throws SQLException { public void batchExecute(String sql, List<List<Map<String, ?>>> batchValues) throws SQLException {
PreparedStatement pstmt = getConn().prepareStatement(sql); PreparedStatement pstmt = getConn().prepareStatement(sql);
for (int i = 0; i < batchValues.size(); i++) { for (int i = 0; i < batchValues.size(); i++) {
List<Map<String, ?>> values = batchValues.get(i); List<Map<String, ?>> values = batchValues.get(i);

View File

@@ -1,12 +1,13 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.support; package com.alibaba.otter.canal.client.adapter.clickhouse.support;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import org.springframework.util.LinkedCaseInsensitiveMap;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import org.springframework.util.LinkedCaseInsensitiveMap;
import com.alibaba.otter.canal.client.adapter.support.Dml;
public class SingleDml { public class SingleDml {
private String destination; private String destination;

View File

@@ -1,12 +1,5 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.support; package com.alibaba.otter.canal.client.adapter.clickhouse.support;
import com.alibaba.druid.DbType;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Reader; import java.io.Reader;
import java.io.StringReader; import java.io.StringReader;
import java.math.BigDecimal; import java.math.BigDecimal;
@@ -16,8 +9,17 @@ import java.util.Collection;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.Map; import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.DbType;
import com.alibaba.otter.canal.client.adapter.clickhouse.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.Util;
public class SyncUtil { public class SyncUtil {
private static final Logger logger = LoggerFactory.getLogger(SyncUtil.class);
private static final Logger logger = LoggerFactory.getLogger(SyncUtil.class);
public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Map<String, Object> data) { public static Map<String, String> getColumnsMap(MappingConfig.DbMapping dbMapping, Map<String, Object> data) {
return getColumnsMap(dbMapping, data.keySet()); return getColumnsMap(dbMapping, data.keySet());

View File

@@ -1,13 +1,13 @@
package com.alibaba.otter.canal.client.adapter.clickhouse; package com.alibaba.otter.canal.client.adapter.clickhouse;
import java.util.*;
import com.alibaba.otter.canal.client.adapter.clickhouse.sync.Common;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import org.junit.Before; import org.junit.Before;
import org.junit.Ignore; import org.junit.Ignore;
import org.junit.Test; import org.junit.Test;
import java.util.*; import com.alibaba.otter.canal.client.adapter.clickhouse.sync.Common;
import com.alibaba.otter.canal.client.adapter.support.Dml;
@Ignore @Ignore
public class ClickHouseBatchSyncServiceTest { public class ClickHouseBatchSyncServiceTest {
@@ -106,4 +106,4 @@ public class ClickHouseBatchSyncServiceTest {
Thread.sleep(1000L); Thread.sleep(1000L);
} }
} }

View File

@@ -1,19 +1,5 @@
package com.alibaba.otter.canal.client.adapter.clickhouse; package com.alibaba.otter.canal.client.adapter.clickhouse;
/**
* @author: Xander
* @date: Created in 2023/11/13 22:27
* @email: zhrunxin33@gmail.com
* @description: Testing thread safe
*/
import ch.qos.logback.classic.Level;
import com.alibaba.otter.canal.client.adapter.clickhouse.sync.Common;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import java.util.*; import java.util.*;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
@@ -21,16 +7,32 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import com.alibaba.otter.canal.client.adapter.clickhouse.sync.Common;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import ch.qos.logback.classic.Level;
/**
* @author: Xander
* @date: Created in 2023/11/13 22:27
* @email: zhrunxin33@gmail.com
* @description: Testing thread safe
*/
@Ignore @Ignore
public class ClickHouseBatchSyncThreadSafeTest { public class ClickHouseBatchSyncThreadSafeTest {
private ClickHouseAdapter clickHouseAdapter; private ClickHouseAdapter clickHouseAdapter;
private ExecutorService executorService; private ExecutorService executorService;
private String[] operations = new String[]{"INSERT", "UPDATE"}; private String[] operations = new String[] { "INSERT", "UPDATE" };
private String[] tables = new String[]{"user", "customer"}; private String[] tables = new String[] { "user", "customer" };
@Before @Before
public void init() { public void init() {
@@ -55,12 +57,12 @@ public class ClickHouseBatchSyncThreadSafeTest {
dml.setTs(new Date().getTime()); dml.setTs(new Date().getTime());
dml.setType(dmlType); dml.setType(dmlType);
dml.setDatabase("mytest"); dml.setDatabase("mytest");
dml.setTable(tables[(int)Math.round(Math.random())]); dml.setTable(tables[(int) Math.round(Math.random())]);
List<Map<String, Object>> dataList = new ArrayList<>(); List<Map<String, Object>> dataList = new ArrayList<>();
Map<String, Object> data = new LinkedHashMap<>(); Map<String, Object> data = new LinkedHashMap<>();
dataList.add(data); dataList.add(data);
data.put("id", cou); data.put("id", cou);
data.put("name", "Eric"+cou); data.put("name", "Eric" + cou);
data.put("role_id", cou); data.put("role_id", cou);
data.put("c_time", new Date()); data.put("c_time", new Date());
data.put("test1", "sdfasdfawe中国asfwef"); data.put("test1", "sdfasdfawe中国asfwef");
@@ -73,7 +75,7 @@ public class ClickHouseBatchSyncThreadSafeTest {
for (Future future : list) { for (Future future : list) {
future.get(); future.get();
} }
Thread.sleep(10000L); // waiting multiple threads execute successfully. Thread.sleep(10000L); // waiting multiple threads execute successfully.
} }
} }

View File

@@ -1,14 +1,14 @@
package com.alibaba.otter.canal.client.adapter.clickhouse; package com.alibaba.otter.canal.client.adapter.clickhouse;
import com.alibaba.druid.pool.DruidDataSource;
import java.sql.SQLException; import java.sql.SQLException;
import com.alibaba.druid.pool.DruidDataSource;
public class TestConstant { public class TestConstant {
public final static String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true"; public final static String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true";
public final static String jdbcUser = "root"; public final static String jdbcUser = "root";
public final static String jdbcPassword = "121212"; public final static String jdbcPassword = "121212";
public final static DruidDataSource dataSource; public final static DruidDataSource dataSource;

View File

@@ -1,21 +1,22 @@
package com.alibaba.otter.canal.client.adapter.clickhouse.sync; package com.alibaba.otter.canal.client.adapter.clickhouse.sync;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter;
import com.alibaba.otter.canal.client.adapter.clickhouse.TestConstant;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
import org.slf4j.LoggerFactory;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter;
import com.alibaba.otter.canal.client.adapter.clickhouse.TestConstant;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
/** /**
* @author: Xander * @author: Xander
* @date: Created in 2023/11/13 0:16 * @date: Created in 2023/11/13 0:16
* @email: zhrunxin33@gmail.com * @email: zhrunxin33@gmail.com @description:
* @description:
*/ */
public class Common { public class Common {
@@ -43,7 +44,8 @@ public class Common {
* @param logLevel * @param logLevel
*/ */
public static void setLogLevel(Level logLevel) { public static void setLogLevel(Level logLevel) {
ch.qos.logback.classic.Logger logger = (ch.qos.logback.classic.Logger)LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME); ch.qos.logback.classic.Logger logger = (ch.qos.logback.classic.Logger) LoggerFactory
.getLogger(Logger.ROOT_LOGGER_NAME);
logger.setLevel(logLevel); logger.setLevel(logLevel);
} }
} }