From 93dac2d335a63e05e726754eae02614ae02b5156 Mon Sep 17 00:00:00 2001 From: "jianghang.loujh" Date: Mon, 4 Dec 2023 09:46:23 +0800 Subject: [PATCH] format clickhouse adapter code --- .../adapter/clickhouse/ClickHouseAdapter.java | 32 ++++++---- .../clickhouse/config/ConfigLoader.java | 17 +++-- .../clickhouse/config/MappingConfig.java | 13 ++-- .../clickhouse/config/MirrorDbConfig.java | 11 +++- .../monitor/ClickHouseConfigMonitor.java | 22 +++++-- .../service/ClickHouseBatchSyncService.java | 28 ++++---- .../service/ClickHouseEtlService.java | 64 ++++++++++--------- .../ClickHouseMirrorDbBatchSyncService.java | 50 +++++++++------ .../clickhouse/support/BatchExecutor.java | 11 ++-- .../adapter/clickhouse/support/SingleDml.java | 7 +- .../adapter/clickhouse/support/SyncUtil.java | 18 +++--- .../ClickHouseBatchSyncServiceTest.java | 8 +-- .../ClickHouseBatchSyncThreadSafeTest.java | 42 ++++++------ .../adapter/clickhouse/TestConstant.java | 10 +-- .../adapter/clickhouse/sync/Common.java | 24 +++---- 15 files changed, 206 insertions(+), 151 deletions(-) diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java index e4a57cbf..fbfe8df3 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseAdapter.java @@ -1,5 +1,15 @@ 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.pool.DruidDataSource; 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.support.SyncUtil; 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 @@ -35,11 +36,15 @@ public class ClickHouseAdapter implements OuterAdapter { private static final Logger logger = LoggerFactory.getLogger(ClickHouseAdapter.class); - private Map 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 clickHouseMapping = new ConcurrentHashMap<>(); - private Map> mappingConfigCache = new ConcurrentHashMap<>(); // Schema -> Table -> MappingConfig + // Schema -> Table -> MappingConfig + private Map> mappingConfigCache = new ConcurrentHashMap<>(); - private Map 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 mirrorDbConfigCache = new ConcurrentHashMap<>(); private DruidDataSource dataSource; @@ -49,7 +54,8 @@ public class ClickHouseAdapter implements OuterAdapter { private Properties envProperties; - private OuterAdapterConfig configuration; // Launch configuration + // Launch configuration + private OuterAdapterConfig configuration; private ClickHouseConfigMonitor clickHouseConfigMonitor; diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java index 15933b5d..fdc311a3 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/ConfigLoader.java @@ -1,19 +1,22 @@ 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.Map; 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表映射配置加载器 * - * @author rewerma 2018-11-07 下午02:41:34 - * @version 1.0.0 + * @author: Xander + * @date: Created in 2023/11/10 22:23 + * @email: zhrunxin33@gmail.com + * @version 1.1.8 */ public class ConfigLoader { diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java index 8d230160..387dac69 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MappingConfig.java @@ -1,16 +1,19 @@ 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.Map; +import org.apache.commons.lang.StringUtils; + +import com.alibaba.otter.canal.client.adapter.support.AdapterConfig; + /** * CLICKHOUSE表映射配置 * - * @author rewerma 2018-11-07 下午02:41:34 - * @version 1.0.0 + * @author: Xander + * @date: Created in 2023/11/10 22:23 + * @email: zhrunxin33@gmail.com + * @version 1.1.8 */ public class MappingConfig implements AdapterConfig { diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java index 5f53b7d5..050cda04 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/config/MirrorDbConfig.java @@ -3,10 +3,17 @@ package com.alibaba.otter.canal.client.adapter.clickhouse.config; import java.util.Map; 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 { - private String fileName; - private MappingConfig mappingConfig; + private String fileName; + private MappingConfig mappingConfig; private Map tableConfig = new ConcurrentHashMap<>(); public static MirrorDbConfig create(String fileName, MappingConfig mappingConfig) { diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java index ff048235..96f02cec 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/monitor/ClickHouseConfigMonitor.java @@ -1,10 +1,8 @@ package com.alibaba.otter.canal.client.adapter.clickhouse.monitor; -import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter; -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; +import java.io.File; +import java.util.Properties; + import org.apache.commons.io.filefilter.FileFilterUtils; import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; 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.LoggerFactory; -import java.io.File; -import java.util.Properties; +import com.alibaba.otter.canal.client.adapter.clickhouse.ClickHouseAdapter; +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 { diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java index 6265d1bf..90b6ffec 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseBatchSyncService.java @@ -1,5 +1,18 @@ 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.fastjson2.JSON; 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.support.Dml; 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 @@ -29,6 +30,7 @@ import java.util.stream.Collectors; * @author: Xander * @date: Created in 2023/11/10 22:23 * @email: zhrunxin33@gmail.com + * @version 1.1.8 */ public class ClickHouseBatchSyncService { @@ -45,7 +47,7 @@ public class ClickHouseBatchSyncService { private BatchExecutor alterExecutors; // Alter Single Executor(update/delete/truncate) private ExecutorService[] executorThreads; // Be initialized once - + private ScheduledExecutorService[] scheduledExecutors; private int threads = 3; // Default parallel thread count diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java index 70e09da4..990900e3 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseEtlService.java @@ -1,5 +1,14 @@ 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.otter.canal.client.adapter.clickhouse.config.MappingConfig; 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.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 操作业务类 * - * @author rewerma @ 2018-11-7 - * @version 1.0.0 + * @author: Xander + * @date: Created in 2023/11/10 22:23 + * @email: zhrunxin33@gmail.com + * @version 1.1.8 */ public class ClickHouseEtlService extends AbstractEtlService { @@ -58,23 +61,23 @@ public class ClickHouseEtlService extends AbstractEtlService { Util.sqlRS(targetDS, "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " LIMIT 1 ", rs -> { - try { + try { - ResultSetMetaData rsd = rs.getMetaData(); - int columnCount = rsd.getColumnCount(); - List columns = new ArrayList<>(); - for (int i = 1; i <= columnCount; i++) { - columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(i)); - columns.add(rsd.getColumnName(i)); + ResultSetMetaData rsd = rs.getMetaData(); + int columnCount = rsd.getColumnCount(); + List columns = new ArrayList<>(); + for (int i = 1; i <= columnCount; i++) { + columnType.put(rsd.getColumnName(i).toLowerCase(), rsd.getColumnType(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 -> { int idx = 1; @@ -86,8 +89,10 @@ public class ClickHouseEtlService extends AbstractEtlService { insertSql.append("INSERT INTO ") .append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType())) .append(" ("); - columnsMap - .forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick).append(targetColumnName).append(backtick).append(",")); + columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick) + .append(targetColumnName) + .append(backtick) + .append(",")); int len = insertSql.length(); insertSql.delete(len - 1, len).append(") VALUES ("); @@ -97,7 +102,7 @@ public class ClickHouseEtlService extends AbstractEtlService { } len = insertSql.length(); insertSql.delete(len - 1, len).append(")"); - logger.info("executeSqlImport sql:{}",insertSql.toString()); + logger.info("executeSqlImport sql:{}", insertSql.toString()); try (Connection connTarget = targetDS.getConnection(); PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) { connTarget.setAutoCommit(false); @@ -110,7 +115,8 @@ public class ClickHouseEtlService extends AbstractEtlService { // 删除数据 Map pkVal = new LinkedHashMap<>(); 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); try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { int k = 1; diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseMirrorDbBatchSyncService.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseMirrorDbBatchSyncService.java index 3af9ae07..74e47a83 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseMirrorDbBatchSyncService.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/service/ClickHouseMirrorDbBatchSyncService.java @@ -1,16 +1,5 @@ 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.Statement; import java.util.ArrayList; @@ -18,26 +7,48 @@ import java.util.LinkedHashMap; import java.util.List; 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镜像库同步操作业务 * - * @author rewerma 2018-12-12 下午011:23 - * @version 1.0.0 + * @author: Xander + * @date: Created in 2023/11/10 22:23 + * @email: zhrunxin33@gmail.com + * @version 1.1.8 */ public class ClickHouseMirrorDbBatchSyncService { private static final Logger logger = LoggerFactory.getLogger(ClickHouseMirrorDbBatchSyncService.class); - private Map mirrorDbConfigCache; // 镜像库配置 + // 镜像库配置 + private Map mirrorDbConfigCache; private DruidDataSource dataSource; - private ClickHouseBatchSyncService clickHouseBatchSyncService; // clickhouseSyncService代理 + // clickhouseSyncService代理 + private ClickHouseBatchSyncService clickHouseBatchSyncService; - public ClickHouseMirrorDbBatchSyncService(Map mirrorDbConfigCache, DruidDataSource dataSource, - Integer threads, Integer batchSize, Long scheduleTime, Map> columnsTypeCache, + public ClickHouseMirrorDbBatchSyncService(Map mirrorDbConfigCache, + DruidDataSource dataSource, Integer threads, Integer batchSize, + Long scheduleTime, Map> columnsTypeCache, boolean skipDupException){ this.mirrorDbConfigCache = mirrorDbConfigCache; 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)); } executeDdl(mirrorDbConfig, dml); - clickHouseBatchSyncService.getColumnsTypeCache().remove(destination + "." + database + "." + dml.getTable()); + clickHouseBatchSyncService.getColumnsTypeCache() + .remove(destination + "." + database + "." + dml.getTable()); mirrorDbConfig.getTableConfig().remove(dml.getTable()); // 删除对应库表配置 } else { // DML diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/BatchExecutor.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/BatchExecutor.java index 008dbb3a..19d5411b 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/BatchExecutor.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/BatchExecutor.java @@ -1,9 +1,5 @@ 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.sql.Connection; import java.sql.PreparedStatement; @@ -13,6 +9,11 @@ import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import javax.sql.DataSource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + /** * sql批量执行器 * @@ -64,7 +65,7 @@ public class BatchExecutor implements Closeable { pstmt.close(); } - public void batchExecute(String sql, List>> batchValues) throws SQLException { + public void batchExecute(String sql, List>> batchValues) throws SQLException { PreparedStatement pstmt = getConn().prepareStatement(sql); for (int i = 0; i < batchValues.size(); i++) { List> values = batchValues.get(i); diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SingleDml.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SingleDml.java index b45ed613..bee6a4d9 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SingleDml.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SingleDml.java @@ -1,12 +1,13 @@ 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.List; import java.util.Map; +import org.springframework.util.LinkedCaseInsensitiveMap; + +import com.alibaba.otter.canal.client.adapter.support.Dml; + public class SingleDml { private String destination; diff --git a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SyncUtil.java b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SyncUtil.java index b1e033b8..ad38a3b4 100644 --- a/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SyncUtil.java +++ b/client-adapter/clickhouse/src/main/java/com/alibaba/otter/canal/client/adapter/clickhouse/support/SyncUtil.java @@ -1,12 +1,5 @@ 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.StringReader; import java.math.BigDecimal; @@ -16,8 +9,17 @@ import java.util.Collection; import java.util.LinkedHashMap; 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 { - private static final Logger logger = LoggerFactory.getLogger(SyncUtil.class); + + private static final Logger logger = LoggerFactory.getLogger(SyncUtil.class); public static Map getColumnsMap(MappingConfig.DbMapping dbMapping, Map data) { return getColumnsMap(dbMapping, data.keySet()); diff --git a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncServiceTest.java b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncServiceTest.java index c6563f3e..2eb339be 100644 --- a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncServiceTest.java +++ b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncServiceTest.java @@ -1,13 +1,13 @@ 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.Ignore; 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 public class ClickHouseBatchSyncServiceTest { @@ -106,4 +106,4 @@ public class ClickHouseBatchSyncServiceTest { Thread.sleep(1000L); } -} \ No newline at end of file +} diff --git a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncThreadSafeTest.java b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncThreadSafeTest.java index 2953b96e..dfb7d0de 100644 --- a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncThreadSafeTest.java +++ b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/ClickHouseBatchSyncThreadSafeTest.java @@ -1,19 +1,5 @@ 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.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -21,16 +7,32 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; 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 public class ClickHouseBatchSyncThreadSafeTest { 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 public void init() { @@ -55,12 +57,12 @@ public class ClickHouseBatchSyncThreadSafeTest { dml.setTs(new Date().getTime()); dml.setType(dmlType); dml.setDatabase("mytest"); - dml.setTable(tables[(int)Math.round(Math.random())]); + dml.setTable(tables[(int) Math.round(Math.random())]); List> dataList = new ArrayList<>(); Map data = new LinkedHashMap<>(); dataList.add(data); data.put("id", cou); - data.put("name", "Eric"+cou); + data.put("name", "Eric" + cou); data.put("role_id", cou); data.put("c_time", new Date()); data.put("test1", "sdfasdfawe中国asfwef"); @@ -73,7 +75,7 @@ public class ClickHouseBatchSyncThreadSafeTest { for (Future future : list) { future.get(); } - Thread.sleep(10000L); // waiting multiple threads execute successfully. + Thread.sleep(10000L); // waiting multiple threads execute successfully. } } diff --git a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/TestConstant.java b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/TestConstant.java index c86cd4a8..5ed59cbd 100644 --- a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/TestConstant.java +++ b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/TestConstant.java @@ -1,14 +1,14 @@ package com.alibaba.otter.canal.client.adapter.clickhouse; -import com.alibaba.druid.pool.DruidDataSource; - import java.sql.SQLException; +import com.alibaba.druid.pool.DruidDataSource; + public class TestConstant { - 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 jdbcPassword = "121212"; + 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 jdbcPassword = "121212"; public final static DruidDataSource dataSource; diff --git a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/sync/Common.java b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/sync/Common.java index 00deac52..6073eb70 100644 --- a/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/sync/Common.java +++ b/client-adapter/clickhouse/src/test/java/com/alibaba/otter/canal/client/adapter/clickhouse/sync/Common.java @@ -1,21 +1,22 @@ 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.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 * @date: Created in 2023/11/13 0:16 - * @email: zhrunxin33@gmail.com - * @description： + * @email: zhrunxin33@gmail.com @description： */ public class Common { @@ -43,7 +44,8 @@ public class Common { * @param 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); } }