From b457ecd387d19c54250dc69cccc5c931cf4ce80d Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 15 Nov 2018 17:09:53 +0800 Subject: [PATCH 1/4] =?UTF-8?q?adapter=20=E4=B8=BB=E9=85=8D=E7=BD=AEapplic?= =?UTF-8?q?ation.yml=E5=8A=A8=E6=80=81=E5=8A=A0=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/adapter/support/Constant.java | 6 ++ .../adapter/support/MappingConfigsLoader.java | 2 +- client-adapter/launcher/pom.xml | 5 + .../launcher/CanalAdapterApplication.java | 6 +- .../launcher/config/RefresherConfig.java | 22 +++++ .../launcher/loader/CanalAdapterLoader.java | 33 ++++++- .../launcher/loader/CanalAdapterService.java | 49 ++++++---- .../monitor/ApplicationRunningMonitor.java | 94 +++++++++++++++++++ .../src/main/resources/application.yml | 1 + 9 files changed, 192 insertions(+), 26 deletions(-) create mode 100644 client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Constant.java create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/RefresherConfig.java create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Constant.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Constant.java new file mode 100644 index 00000000..9a75f765 --- /dev/null +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Constant.java @@ -0,0 +1,6 @@ +package com.alibaba.otter.canal.client.adapter.support; + +public class Constant { + + public static final String CONF_DIR = "conf"; +} diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java index 711a52b6..c437731b 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MappingConfigsLoader.java @@ -15,7 +15,7 @@ public class MappingConfigsLoader { Map configContentMap = new HashMap<>(); // 先取本地文件，再取类路径 - File configDir = new File("../conf/" + name); + File configDir = new File(".." + File.separator + Constant.CONF_DIR + File.separator + name); if (!configDir.exists()) { URL url = MappingConfigsLoader.class.getClassLoader().getResource(""); if (url != null) { diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index df31b8c8..6a313934 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -49,6 +49,11 @@ spring-boot-configuration-processor provided + + org.springframework.cloud + spring-cloud-context + 2.0.0.RELEASE + org.apache.curator curator-recipes diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java index 99140b0c..83f4b350 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.adapter.launcher; +import org.springframework.boot.Banner; +import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.builder.SpringApplicationBuilder; @@ -13,6 +15,8 @@ import org.springframework.boot.builder.SpringApplicationBuilder; public class CanalAdapterApplication { public static void main(String[] args) { - new SpringApplicationBuilder(CanalAdapterApplication.class).run(args); + SpringApplication application = new SpringApplication(CanalAdapterApplication.class); + application.setBannerMode(Banner.Mode.OFF); + application.run(args); } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/RefresherConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/RefresherConfig.java new file mode 100644 index 00000000..127b7ceb --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/RefresherConfig.java @@ -0,0 +1,22 @@ +package com.alibaba.otter.canal.adapter.launcher.config; + +import org.springframework.cloud.context.refresh.ContextRefresher; +import org.springframework.cloud.context.scope.refresh.RefreshScope; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RefresherConfig { + + @Bean + public RefreshScope refreshScope() { + return new RefreshScope(); + } + + @Bean + public ContextRefresher contextRefresher(ConfigurableApplicationContext configurableApplicationContext, + RefreshScope refreshScope) { + return new ContextRefresher(configurableApplicationContext, refreshScope); + } +} diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index e2d9bea8..900d1207 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -6,8 +6,10 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; @@ -136,7 +138,6 @@ public class CanalAdapterLoader { logger.info("Load canal adapter: {} succeed", config.getName()); } catch (Exception e) { logger.error("Load canal adapter: {} failed", config.getName(), e); - System.exit(0); } } @@ -144,19 +145,41 @@ public class CanalAdapterLoader { * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁 */ public void destroy() { - if (canalWorkers.size() > 0) { + if (!canalWorkers.isEmpty()) { ExecutorService stopExecutorService = Executors.newFixedThreadPool(canalWorkers.size()); + List> futures = new ArrayList<>(); for (CanalAdapterWorker canalAdapterWorker : canalWorkers.values()) { - stopExecutorService.submit(canalAdapterWorker::stop); + futures.add(stopExecutorService.submit(() -> { + canalAdapterWorker.stop(); + return true; + })); } + futures.forEach(future -> { + try { + future.get(); + } catch (Exception e) { + // ignore + } + }); stopExecutorService.shutdown(); } - if (canalMQWorker.size() > 0) { + if (!canalMQWorker.isEmpty()) { ExecutorService stopMQWorkerService = Executors.newFixedThreadPool(canalMQWorker.size()); + List> futures = new ArrayList<>(); for (AbstractCanalAdapterWorker canalAdapterMQWorker : canalMQWorker.values()) { - stopMQWorkerService.submit(canalAdapterMQWorker::stop); + futures.add(stopMQWorkerService.submit(() -> { + canalAdapterMQWorker.stop(); + return true; + })); } + futures.forEach(future -> { + try { + future.get(); + } catch (Exception e) { + // ignore + } + }); stopMQWorkerService.shutdown(); } logger.info("All canal adapters destroyed"); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java index 15dafa79..79eb0be1 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterService.java @@ -4,14 +4,15 @@ import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import javax.annotation.Resource; -import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; -import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.cloud.context.config.annotation.RefreshScope; import org.springframework.stereotype.Component; import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; import com.alibaba.otter.canal.adapter.launcher.config.AdapterCanalConfig; +import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; /** @@ -21,42 +22,51 @@ import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; * @version 1.0.0 */ @Component +@RefreshScope public class CanalAdapterService { - private static final Logger logger = LoggerFactory.getLogger(CanalAdapterService.class); + private static final Logger logger = LoggerFactory.getLogger(CanalAdapterService.class); - private static CanalAdapterLoader adapterLoader; + private CanalAdapterLoader adapterLoader; @Resource - private AdapterCanalConfig adapterCanalConfig; + private AdapterCanalConfig adapterCanalConfig; // 注入bean保证优先注册 @Resource - private SpringContext springContext; + private SpringContext springContext; @Resource - private SyncSwitch syncSwitch; + private SyncSwitch syncSwitch; + + private volatile boolean running = false; @PostConstruct - public void init() { - if (adapterLoader == null) { - try { - logger.info("## start the canal client adapters."); - adapterLoader = new CanalAdapterLoader(adapterCanalConfig); - adapterLoader.init(); - logger.info("## the canal client adapters are running now ......"); - } catch (Throwable e) { - logger.error("## something goes wrong when starting up the canal client adapters:", e); - System.exit(0); - } + public synchronized void init() { + if (running) { + return; + } + try { + logger.info("## start the canal client adapters."); + adapterLoader = new CanalAdapterLoader(adapterCanalConfig); + adapterLoader.init(); + running = true; + logger.info("## the canal client adapters are running now ......"); + } catch (Exception e) { + logger.error("## something goes wrong when starting up the canal client adapters:", e); } } @PreDestroy - public void destroy() { + public synchronized void destroy() { + if (!running) { + return; + } try { + running = false; logger.info("## stop the canal client adapters"); if (adapterLoader != null) { adapterLoader.destroy(); + adapterLoader = null; } for (DruidDataSource druidDataSource : DatasourceConfig.DATA_SOURCES.values()) { try { @@ -65,6 +75,7 @@ public class CanalAdapterService { logger.error(e.getMessage(), e); } } + DatasourceConfig.DATA_SOURCES.clear(); } catch (Throwable e) { logger.warn("## something goes wrong when stopping canal client adapters:", e); } finally { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java new file mode 100644 index 00000000..b1be1b76 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java @@ -0,0 +1,94 @@ +package com.alibaba.otter.canal.adapter.launcher.monitor; + +import java.io.File; +import java.io.FileReader; +import java.net.URL; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import javax.annotation.Resource; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.cloud.context.refresh.ContextRefresher; +import org.springframework.stereotype.Component; +import org.yaml.snakeyaml.Yaml; + +import com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterService; +import com.alibaba.otter.canal.client.adapter.support.Constant; + +@Component +public class ApplicationRunningMonitor { + + private static final Logger logger = LoggerFactory.getLogger(ApplicationRunningMonitor.class); + + @Resource + private ContextRefresher contextRefresher; + + @Resource + private CanalAdapterService canalAdapterService; + + private ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1); + + private static volatile long applicationLastModify = -1L; + + @PostConstruct + public void init() { + URL url = this.getClass().getClassLoader().getResource(""); + String path; + if (url != null) { + path = url.getPath(); + } else { + path = new File("").getAbsolutePath(); + } + File file = null; + if (path != null) { + file = new File(path + ".." + File.separator + Constant.CONF_DIR + File.separator + "application.yml"); + if (!file.exists()) { + file = new File(path + "application.yml"); + } + } + if (file == null || !file.exists()) { + throw new RuntimeException("application.yml config file not found."); + } + File configFile = file; + scheduled.scheduleWithFixedDelay(() -> { + if (applicationLastModify == -1L) { + applicationLastModify = configFile.lastModified(); + } else { + if (configFile.lastModified() != applicationLastModify) { + applicationLastModify = configFile.lastModified(); + + try { + // 检查yml格式 + new Yaml().loadAs(new FileReader(configFile), Map.class); + + canalAdapterService.destroy(); + + // refresh context + contextRefresher.refresh(); + + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + // ignore + } + canalAdapterService.init(); + logger.info("## adapter application config reloaded."); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } + }, 5, 5, TimeUnit.SECONDS); + } + + @PreDestroy + public void destroy() { + scheduled.shutdown(); + } +} diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index e5d4f507..f862bb0e 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -2,6 +2,7 @@ server: port: 8081 logging: level: +# org.springframework: WARN com.alibaba.otter.canal.client.adapter.hbase: DEBUG com.alibaba.otter.canal.client.adapter.es: DEBUG com.alibaba.otter.canal.client.adapter.rdb: DEBUG From 7ee7ba0e05ae2c5f294670b46a5bcb92f34e5f0b Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 15 Nov 2018 22:47:08 +0800 Subject: [PATCH 2/4] =?UTF-8?q?rdb=20=E6=89=B9=E9=87=8F=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/adapter/support/Util.java | 30 +++ .../client/adapter/hbase/HbaseAdapter.java | 5 +- .../monitor/ApplicationRunningMonitor.java | 112 ++++----- .../src/main/resources/application.yml | 22 +- client-adapter/rdb/pom.xml | 10 +- .../canal/client/adapter/rdb/RdbAdapter.java | 63 ++++- .../adapter/rdb/service/BatchExecutor.java | 48 ++-- .../adapter/rdb/service/RdbSyncService.java | 238 +++++++++++------- 8 files changed, 338 insertions(+), 190 deletions(-) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java index ac2c32c3..e3c8032c 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Util.java @@ -1,5 +1,7 @@ package com.alibaba.otter.canal.client.adapter.support; +import java.io.File; +import java.net.URL; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; @@ -9,6 +11,7 @@ import java.util.function.Function; import javax.sql.DataSource; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -87,4 +90,31 @@ public class Util { } } } + + public static File getConfDirPath() { + return getConfDirPath(""); + } + + public static File getConfDirPath(String subConf) { + URL url = Util.class.getClassLoader().getResource(""); + String path; + if (url != null) { + path = url.getPath(); + } else { + path = new File("").getAbsolutePath(); + } + File file = null; + if (path != null) { + file = new File( + path + ".." + File.separator + Constant.CONF_DIR + File.separator + StringUtils.trimToEmpty(subConf)); + if (!file.exists()) { + file = new File(path + StringUtils.trimToEmpty(subConf)); + } + } + if (file == null || !file.exists()) { + throw new RuntimeException("Config dir not found."); + } + + return file; + } } diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index a64ca3dc..9c0395f3 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -5,6 +5,7 @@ import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import javax.sql.DataSource; @@ -37,8 +38,8 @@ public class HbaseAdapter implements OuterAdapter { private static Logger logger = LoggerFactory.getLogger(HbaseAdapter.class); - private Map hbaseMapping = new HashMap<>(); // 文件名对应配置 - private Map mappingConfigCache = new HashMap<>(); // 库名-表名对应配置 + private Map hbaseMapping = new ConcurrentHashMap<>(); // 文件名对应配置 + private Map mappingConfigCache = new ConcurrentHashMap<>(); // 库名-表名对应配置 private Connection conn; private HbaseSyncService hbaseSyncService; diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java index b1be1b76..0397464d 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/ApplicationRunningMonitor.java @@ -2,16 +2,16 @@ package com.alibaba.otter.canal.adapter.launcher.monitor; import java.io.File; import java.io.FileReader; -import java.net.URL; import java.util.Map; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import javax.annotation.Resource; +import org.apache.commons.io.filefilter.FileFilterUtils; +import org.apache.commons.io.monitor.FileAlterationListenerAdaptor; +import org.apache.commons.io.monitor.FileAlterationMonitor; +import org.apache.commons.io.monitor.FileAlterationObserver; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.cloud.context.refresh.ContextRefresher; @@ -19,76 +19,72 @@ import org.springframework.stereotype.Component; import org.yaml.snakeyaml.Yaml; import com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterService; -import com.alibaba.otter.canal.client.adapter.support.Constant; +import com.alibaba.otter.canal.client.adapter.support.Util; @Component public class ApplicationRunningMonitor { - private static final Logger logger = LoggerFactory.getLogger(ApplicationRunningMonitor.class); + private static final Logger logger = LoggerFactory.getLogger(ApplicationRunningMonitor.class); @Resource - private ContextRefresher contextRefresher; + private ContextRefresher contextRefresher; @Resource - private CanalAdapterService canalAdapterService; + private CanalAdapterService canalAdapterService; - private ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1); - - private static volatile long applicationLastModify = -1L; + private FileAlterationMonitor fileMonitor; @PostConstruct public void init() { - URL url = this.getClass().getClassLoader().getResource(""); - String path; - if (url != null) { - path = url.getPath(); - } else { - path = new File("").getAbsolutePath(); + File confDir = Util.getConfDirPath(); + try { + FileAlterationObserver observer = new FileAlterationObserver(confDir, + FileFilterUtils.and(FileFilterUtils.fileFileFilter(), + FileFilterUtils.prefixFileFilter("application"), + FileFilterUtils.suffixFileFilter("yml"))); + FileListener listener = new FileListener(); + observer.addListener(listener); + fileMonitor = new FileAlterationMonitor(3000, observer); + fileMonitor.start(); + + } catch (Exception e) { + logger.error(e.getMessage(), e); } - File file = null; - if (path != null) { - file = new File(path + ".." + File.separator + Constant.CONF_DIR + File.separator + "application.yml"); - if (!file.exists()) { - file = new File(path + "application.yml"); - } - } - if (file == null || !file.exists()) { - throw new RuntimeException("application.yml config file not found."); - } - File configFile = file; - scheduled.scheduleWithFixedDelay(() -> { - if (applicationLastModify == -1L) { - applicationLastModify = configFile.lastModified(); - } else { - if (configFile.lastModified() != applicationLastModify) { - applicationLastModify = configFile.lastModified(); - - try { - // 检查yml格式 - new Yaml().loadAs(new FileReader(configFile), Map.class); - - canalAdapterService.destroy(); - - // refresh context - contextRefresher.refresh(); - - try { - Thread.sleep(2000); - } catch (InterruptedException e) { - // ignore - } - canalAdapterService.init(); - logger.info("## adapter application config reloaded."); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - } - }, 5, 5, TimeUnit.SECONDS); } @PreDestroy public void destroy() { - scheduled.shutdown(); + try { + fileMonitor.stop(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + private class FileListener extends FileAlterationListenerAdaptor { + + @Override + public void onFileChange(File file) { + super.onFileChange(file); + try { + // 检查yml格式 + new Yaml().loadAs(new FileReader(file), Map.class); + + canalAdapterService.destroy(); + + // refresh context + contextRefresher.refresh(); + + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + // ignore + } + canalAdapterService.init(); + logger.info("## adapter application config reloaded."); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } } } diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index f862bb0e..1d78aead 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -2,7 +2,7 @@ server: port: 8081 logging: level: -# org.springframework: WARN + org.springframework: WARN com.alibaba.otter.canal.client.adapter.hbase: DEBUG com.alibaba.otter.canal.client.adapter.es: DEBUG com.alibaba.otter.canal.client.adapter.rdb: DEBUG @@ -28,16 +28,16 @@ canal.conf: - instance: example groups: - outAdapters: - - name: logger -# - name: rdb -# key: oracle1 -# properties: -# jdbc.driverClassName: oracle.jdbc.OracleDriver -# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE -# jdbc.username: mytest -# jdbc.password: m121212 -# threads: 5 -# commitSize: 3000 +# - name: logger + - name: rdb + key: oracle1 + properties: + jdbc.driverClassName: oracle.jdbc.OracleDriver + jdbc.url: jdbc:oracle:thin:@localhost:49161:XE + jdbc.username: mytest + jdbc.password: m121212 + threads: 5 + commitSize: 5000 # - name: rdb # key: postgres1 # properties: diff --git a/client-adapter/rdb/pom.xml b/client-adapter/rdb/pom.xml index 120aa5e1..e467cfca 100644 --- a/client-adapter/rdb/pom.xml +++ b/client-adapter/rdb/pom.xml @@ -39,12 +39,12 @@ - - - - + + com.oracle + ojdbc6 + 11.2.0.4.0-atlassian-hosted - + diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index 68f740a7..729b8959 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -3,6 +3,13 @@ package com.alibaba.otter.canal.client.adapter.rdb; import java.sql.Connection; import java.sql.SQLException; import java.util.*; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import javax.sql.DataSource; @@ -30,6 +37,8 @@ public class RdbAdapter implements OuterAdapter { private RdbSyncService rdbSyncService; + private int commitSize = 3000; + @Override public void init(OuterAdapterConfig configuration) { Map rdbMappingTmp = ConfigLoader.load(); @@ -69,19 +78,58 @@ public class RdbAdapter implements OuterAdapter { String threads = properties.get("threads"); String commitSize = properties.get("commitSize"); - rdbSyncService = new RdbSyncService(commitSize != null ? Integer.valueOf(commitSize) : null, + if (commitSize != null) { + this.commitSize = Integer.valueOf(commitSize); + } + rdbSyncService = new RdbSyncService(this.commitSize, threads != null ? Integer.valueOf(threads) : null, dataSource); } + private AtomicInteger batchRowNum = new AtomicInteger(0); + private List dmlList = Collections.synchronizedList(new ArrayList<>()); + private Lock syncLock = new ReentrantLock(); + private Condition condition = syncLock.newCondition(); + private ExecutorService executor = Executors.newFixedThreadPool(1); + @Override public void sync(Dml dml) { - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - String table = dml.getTable(); - MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); + boolean first = batchRowNum.get() == 0; + int currentSize = batchRowNum.addAndGet(dml.getData().size()); + dmlList.add(dml); - rdbSyncService.sync(config, dml); + if (first) { + // 开启超时判断 + executor.submit(() -> { + try { + syncLock.lock(); + if (!condition.await(5, TimeUnit.SECONDS)) { + // 批量超时 + sync(); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } finally { + syncLock.unlock(); + } + }); + } + + if (currentSize > commitSize) { + sync(); + } + } + + private void sync() { + try { + syncLock.lock(); + rdbSyncService.sync(mappingConfigCache, dmlList); + batchRowNum.set(0); + dmlList.clear(); + condition.signal(); + } finally { + syncLock.unlock(); + } } @Override @@ -178,9 +226,12 @@ public class RdbAdapter implements OuterAdapter { @Override public void destroy() { + executor.shutdown(); + if (rdbSyncService != null) { rdbSyncService.close(); } + if (dataSource != null) { dataSource.close(); } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java index ef0ec222..5e8c65e8 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java @@ -71,32 +71,32 @@ public class BatchExecutor { } catch (SQLException e) { logger.error(e.getMessage(), e); } - int i = idx.incrementAndGet(); - - // 批次的第一次执行设置延时 - if (i == 1) { - executor.submit(() -> { - try { - commitLock.lock(); - conn.commit(); //直接提交一次 - if (!condition.await(5, TimeUnit.SECONDS)) { - // 超时提交 - commit(); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } finally { - commitLock.unlock(); - } - }); - } - - if (i == commitSize) { - commit(); - } + // int i = idx.incrementAndGet(); + // + // // 批次的第一次执行设置延时 + // if (i == 1) { + // executor.submit(() -> { + // try { + // commitLock.lock(); + // conn.commit(); //直接提交一次 + // if (!condition.await(5, TimeUnit.SECONDS)) { + // // 超时提交 + // commit(); + // } + // } catch (Exception e) { + // logger.error(e.getMessage(), e); + // } finally { + // commitLock.unlock(); + // } + // }); + // } + // + // if (i == commitSize) { + // commit(); + // } } - private void commit() { + public void commit() { try { commitLock.lock(); conn.commit(); diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 0eff385b..04bf79da 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -5,17 +5,13 @@ import java.io.StringReader; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.sql.*; -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ThreadPoolExecutor; +import java.sql.Date; +import java.util.*; +import java.util.concurrent.*; import javax.sql.DataSource; +import org.apache.commons.lang.StringUtils; import org.joda.time.DateTime; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -66,7 +62,81 @@ public class RdbSyncService { } } - public void sync(MappingConfig config, List dmlList) { + @SuppressWarnings("unchecked") + public void sync(Map mappingConfigCache, List dmlList) { + try { + List>[] dmlPartition = new ArrayList[threads]; + for (int i = 0; i < threads; i++) { + dmlPartition[i] = new ArrayList<>(); + } + // 根据hash拆分 + dmlList.forEach(dml -> { + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); + + Dml[] dmls = new Dml[threads]; + for (int i = 0; i < threads; i++) { + Dml dmlTmp = new Dml(); + dmlTmp.setDestination(dml.getDestination()); + dmlTmp.setDatabase(dml.getDatabase()); + dmlTmp.setTable(dml.getTable()); + dmlTmp.setType(dml.getType()); + dmlTmp.setTs(dml.getTs()); + dmlTmp.setEs(dml.getEs()); + dmlTmp.setSql(dml.getSql()); + dmlTmp.setData(new ArrayList<>()); + dmlTmp.setOld(new ArrayList<>()); + dmls[i] = dmlTmp; + } + int idx = 0; + for (Map data : dml.getData()) { + int hash; + if (config.getConcurrent()) { + hash = pkHash(config.getDbMapping(), data, threads); + } else { + hash = Math.abs(Math.abs(config.getDbMapping().getTargetTable().hashCode()) % threads); + } + Dml dmlTmp = dmls[hash]; + dmlTmp.getData().add(data); + if (dml.getOld() != null) { + dmlTmp.getOld().add(dml.getOld().get(idx)); + } + + idx++; + } + for (int i = 0; i < threads; i++) { + Map item = new HashMap<>(); + item.put("dml", dmls[i]); + item.put("config", config); + dmlPartition[i].add(item); + } + + }); + List> futures = new ArrayList<>(); + for (int i = 0; i < threads; i++) { + int j = i; + futures.add(threadExecutors[i].submit(() -> { + dmlPartition[j].forEach(item -> { + MappingConfig config = (MappingConfig) item.get("config"); + Dml dml = (Dml) item.get("dml"); + sync(config, dml); + }); + batchExecutors[j].commit(); + return true; + })); + } + for (int i = 0; i < threads; i++) { + try { + futures.get(i).get(); + } catch (Exception e) { + // ignore + } + } + } catch (Exception e) { + logger.error("Error rdb sync for batch", e); + } } public void sync(MappingConfig config, Dml dml) { @@ -134,32 +204,32 @@ public class RdbSyncService { } else { hash = pkHash(dbMapping, d, threads); } - ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; - checkQueue(tpe); - tpe.submit(() -> { - try { - BatchExecutor batchExecutor = batchExecutors[hash]; - List> values = new ArrayList<>(); + // ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; + // checkQueue(tpe); + // tpe.submit(() -> { + try { + BatchExecutor batchExecutor = batchExecutors[hash]; + List> values = new ArrayList<>(); - for (Map.Entry entry : columnsMap.entrySet()) { - String targetColumnName = entry.getKey(); - String srcColumnName = entry.getValue(); - if (srcColumnName == null) { - srcColumnName = targetColumnName; - } - - Integer type = ctype.get(targetColumnName.toLowerCase()); - if (type == null) { - throw new RuntimeException("No column: " + targetColumnName + " found in target db"); - } - Object value = d.get(srcColumnName); - BatchExecutor.setValue(values, type, value); + for (Map.Entry entry : columnsMap.entrySet()) { + String targetColumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetColumnName; } - batchExecutor.execute(sql, values); - } catch (Exception e) { - logger.error(e.getMessage(), e); + + Integer type = ctype.get(targetColumnName.toLowerCase()); + if (type == null) { + throw new RuntimeException("No column: " + targetColumnName + " found in target db"); + } + Object value = d.get(srcColumnName); + BatchExecutor.setValue(values, type, value); } - }); + batchExecutor.execute(sql, values); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + // }); } } @@ -202,46 +272,46 @@ public class RdbSyncService { hash = pkHash(dbMapping, d, o, threads); } - ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; - checkQueue(tpe); - tpe.submit(() -> { - try { - BatchExecutor batchExecutor = batchExecutors[hash]; + // ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; + // checkQueue(tpe); + // tpe.submit(() -> { + try { + BatchExecutor batchExecutor = batchExecutors[hash]; - StringBuilder updateSql = new StringBuilder(); - updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); + StringBuilder updateSql = new StringBuilder(); + updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); - List> values = new ArrayList<>(); - for (String srcColumnName : o.keySet()) { - List targetColumnNames = new ArrayList<>(); - columnsMap.forEach((targetColumn, srcColumn) -> { - if (srcColumnName.toLowerCase().equals(srcColumn)) { - targetColumnNames.add(targetColumn); - } - }); - if (!targetColumnNames.isEmpty()) { - for (String targetColumnName : targetColumnNames) { - updateSql.append(targetColumnName).append("=?, "); - Integer type = ctype.get(targetColumnName.toLowerCase()); - BatchExecutor.setValue(values, type, d.get(srcColumnName)); - } + List> values = new ArrayList<>(); + for (String srcColumnName : o.keySet()) { + List targetColumnNames = new ArrayList<>(); + columnsMap.forEach((targetColumn, srcColumn) -> { + if (srcColumnName.toLowerCase().equals(srcColumn)) { + targetColumnNames.add(targetColumn); + } + }); + if (!targetColumnNames.isEmpty()) { + for (String targetColumnName : targetColumnNames) { + updateSql.append(targetColumnName).append("=?, "); + Integer type = ctype.get(targetColumnName.toLowerCase()); + BatchExecutor.setValue(values, type, d.get(srcColumnName)); } } - int len = updateSql.length(); - updateSql.delete(len - 2, len).append(" WHERE "); - - // 拼接主键 - appendCondition(dbMapping, updateSql, ctype, values, d, o); - - batchExecutor.execute(updateSql.toString(), values); - - if (logger.isDebugEnabled()) { - logger.debug("Execute sql: {}", updateSql); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); } - }); + int len = updateSql.length(); + updateSql.delete(len - 2, len).append(" WHERE "); + + // 拼接主键 + appendCondition(dbMapping, updateSql, ctype, values, d, o); + + batchExecutor.execute(updateSql.toString(), values); + + if (logger.isTraceEnabled()) { + logger.trace("Execute sql: {}", updateSql); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + // }); idx++; } @@ -276,27 +346,27 @@ public class RdbSyncService { hash = pkHash(dbMapping, d, threads); } - ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; - checkQueue(tpe); - tpe.submit(() -> { - try { - BatchExecutor batchExecutor = batchExecutors[hash]; - StringBuilder sql = new StringBuilder(); - sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); + // ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; + // checkQueue(tpe); + // tpe.submit(() -> { + try { + BatchExecutor batchExecutor = batchExecutors[hash]; + StringBuilder sql = new StringBuilder(); + sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); - List> values = new ArrayList<>(); + List> values = new ArrayList<>(); - // 拼接主键 - appendCondition(dbMapping, sql, ctype, values, d); + // 拼接主键 + appendCondition(dbMapping, sql, ctype, values, d); - batchExecutor.execute(sql.toString(), values); - if (logger.isDebugEnabled()) { - logger.debug("Execute sql: {}", sql); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); + batchExecutor.execute(sql.toString(), values); + if (logger.isTraceEnabled()) { + logger.trace("Execute sql: {}", sql); } - }); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + // }); } } From bc056288d4454aca2616dd8f74c03e6c7bee9a74 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 16 Nov 2018 13:12:50 +0800 Subject: [PATCH 3/4] =?UTF-8?q?rdb=20=E6=89=B9=E9=87=8F=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application.yml | 20 +- .../canal/client/adapter/rdb/RdbAdapter.java | 79 ++-- .../adapter/rdb/service/BatchExecutor.java | 29 +- .../adapter/rdb/service/RdbSyncService.java | 365 ++++++------------ .../client/adapter/rdb/support/SimpleDml.java | 102 +++++ 5 files changed, 282 insertions(+), 313 deletions(-) create mode 100644 client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 1d78aead..3df3f4be 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -28,16 +28,16 @@ canal.conf: - instance: example groups: - outAdapters: -# - name: logger - - name: rdb - key: oracle1 - properties: - jdbc.driverClassName: oracle.jdbc.OracleDriver - jdbc.url: jdbc:oracle:thin:@localhost:49161:XE - jdbc.username: mytest - jdbc.password: m121212 - threads: 5 - commitSize: 5000 + - name: logger +# - name: rdb +# key: oracle1 +# properties: +# jdbc.driverClassName: oracle.jdbc.OracleDriver +# jdbc.url: jdbc:oracle:thin:@localhost:49161:XE +# jdbc.username: mytest +# jdbc.password: m121212 +# threads: 5 +# commitSize: 5000 # - name: rdb # key: postgres1 # properties: diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java index 729b8959..4e7e2504 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java @@ -5,9 +5,6 @@ import java.sql.SQLException; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -18,11 +15,14 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.rdb.config.ConfigLoader; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService; import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService; +import com.alibaba.otter.canal.client.adapter.rdb.support.SimpleDml; import com.alibaba.otter.canal.client.adapter.support.*; @SPI("rdb") @@ -30,8 +30,8 @@ public class RdbAdapter implements OuterAdapter { private static Logger logger = LoggerFactory.getLogger(RdbAdapter.class); - private Map rdbMapping = new HashMap<>(); // 文件名对应配置 - private Map mappingConfigCache = new HashMap<>(); // 库名-表名对应配置 + private Map rdbMapping = new HashMap<>(); // 文件名对应配置 + private Map mappingConfigCache = new HashMap<>(); // 库名-表名对应配置 private DruidDataSource dataSource; @@ -39,6 +39,12 @@ public class RdbAdapter implements OuterAdapter { private int commitSize = 3000; + private volatile boolean running = false; + + private List dmlList = Collections.synchronizedList(new ArrayList<>()); + private Lock syncLock = new ReentrantLock(); + private ExecutorService executor = Executors.newFixedThreadPool(1); + @Override public void init(OuterAdapterConfig configuration) { Map rdbMappingTmp = ConfigLoader.load(); @@ -81,52 +87,54 @@ public class RdbAdapter implements OuterAdapter { if (commitSize != null) { this.commitSize = Integer.valueOf(commitSize); } - rdbSyncService = new RdbSyncService(this.commitSize, - threads != null ? Integer.valueOf(threads) : null, - dataSource); - } + rdbSyncService = new RdbSyncService(threads != null ? Integer.valueOf(threads) : null, dataSource); - private AtomicInteger batchRowNum = new AtomicInteger(0); - private List dmlList = Collections.synchronizedList(new ArrayList<>()); - private Lock syncLock = new ReentrantLock(); - private Condition condition = syncLock.newCondition(); - private ExecutorService executor = Executors.newFixedThreadPool(1); + running = true; - @Override - public void sync(Dml dml) { - boolean first = batchRowNum.get() == 0; - int currentSize = batchRowNum.addAndGet(dml.getData().size()); - dmlList.add(dml); - - if (first) { - // 开启超时判断 - executor.submit(() -> { + executor.submit(() -> { + while (running) { try { - syncLock.lock(); - if (!condition.await(5, TimeUnit.SECONDS)) { - // 批量超时 + int size1 = dmlList.size(); + Thread.sleep(3000); + int size2 = dmlList.size(); + if (size1 == size2) { + // 超时提交 sync(); } } catch (Exception e) { logger.error(e.getMessage(), e); - } finally { - syncLock.unlock(); } - }); + } + }); + } + + @Override + public void sync(Dml dml) { + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); + + List simpleDmlList = SimpleDml.dml2SimpleDml(dml, config); + + dmlList.addAll(simpleDmlList); + + if (dmlList.size() > commitSize) { + sync(); } - if (currentSize > commitSize) { - sync(); + if (logger.isDebugEnabled()) { + logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); } } private void sync() { try { syncLock.lock(); - rdbSyncService.sync(mappingConfigCache, dmlList); - batchRowNum.set(0); - dmlList.clear(); - condition.signal(); + if (!dmlList.isEmpty()) { + rdbSyncService.sync(dmlList); + dmlList.clear(); + } } finally { syncLock.unlock(); } @@ -226,6 +234,7 @@ public class RdbAdapter implements OuterAdapter { @Override public void destroy() { + running = false; executor.shutdown(); if (rdbSyncService != null) { diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java index 5e8c65e8..3e3d442c 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/BatchExecutor.java @@ -23,18 +23,14 @@ public class BatchExecutor { private Integer key; private Connection conn; - private int commitSize = 3000; private AtomicInteger idx = new AtomicInteger(0); private ExecutorService executor = Executors.newFixedThreadPool(1); private Lock commitLock = new ReentrantLock(); private Condition condition = commitLock.newCondition(); - public BatchExecutor(Integer key, Connection conn, Integer commitSize){ + public BatchExecutor(Integer key, Connection conn){ this.key = key; this.conn = conn; - if (commitSize != null) { - this.commitSize = commitSize; - } try { this.conn.setAutoCommit(false); @@ -71,29 +67,6 @@ public class BatchExecutor { } catch (SQLException e) { logger.error(e.getMessage(), e); } - // int i = idx.incrementAndGet(); - // - // // 批次的第一次执行设置延时 - // if (i == 1) { - // executor.submit(() -> { - // try { - // commitLock.lock(); - // conn.commit(); //直接提交一次 - // if (!condition.await(5, TimeUnit.SECONDS)) { - // // 超时提交 - // commit(); - // } - // } catch (Exception e) { - // logger.error(e.getMessage(), e); - // } finally { - // commitLock.unlock(); - // } - // }); - // } - // - // if (i == commitSize) { - // commit(); - // } } public void commit() { diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java index 04bf79da..9467b8b9 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java @@ -5,23 +5,22 @@ import java.io.StringReader; import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.sql.*; -import java.sql.Date; -import java.util.*; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.concurrent.*; import javax.sql.DataSource; -import org.apache.commons.lang.StringUtils; import org.joda.time.DateTime; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping; +import com.alibaba.otter.canal.client.adapter.rdb.support.SimpleDml; import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; -import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.Util; /** @@ -42,7 +41,7 @@ public class RdbSyncService { private ExecutorService[] threadExecutors; - public RdbSyncService(Integer commitSize, Integer threads, DataSource dataSource){ + public RdbSyncService(Integer threads, DataSource dataSource){ try { if (threads != null && threads > 1 && threads <= 10) { this.threads = threads; @@ -51,7 +50,7 @@ public class RdbSyncService { for (int i = 0; i < this.threads; i++) { Connection conn = dataSource.getConnection(); conn.setAutoCommit(false); - this.batchExecutors[i] = new BatchExecutor(i, conn, commitSize); + this.batchExecutors[i] = new BatchExecutor(i, conn); } threadExecutors = new ExecutorService[this.threads]; for (int i = 0; i < this.threads; i++) { @@ -63,96 +62,60 @@ public class RdbSyncService { } @SuppressWarnings("unchecked") - public void sync(Map mappingConfigCache, List dmlList) { - try { - List>[] dmlPartition = new ArrayList[threads]; - for (int i = 0; i < threads; i++) { - dmlPartition[i] = new ArrayList<>(); + private List[] simpleDmls2Partition(List simpleDmlList) { + List[] simpleDmlPartition = new ArrayList[threads]; + for (int i = 0; i < threads; i++) { + simpleDmlPartition[i] = new ArrayList<>(); + } + simpleDmlList.forEach(simpleDml -> { + int hash; + if (simpleDml.getConfig().getConcurrent()) { + hash = pkHash(simpleDml.getConfig().getDbMapping(), simpleDml.getData(), threads); + } else { + hash = Math.abs(Math.abs(simpleDml.getConfig().getDbMapping().getTargetTable().hashCode()) % threads); } - // 根据hash拆分 - dmlList.forEach(dml -> { - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - String table = dml.getTable(); - MappingConfig config = mappingConfigCache.get(destination + "." + database + "." + table); + simpleDmlPartition[hash].add(simpleDml); + }); + return simpleDmlPartition; + } - Dml[] dmls = new Dml[threads]; - for (int i = 0; i < threads; i++) { - Dml dmlTmp = new Dml(); - dmlTmp.setDestination(dml.getDestination()); - dmlTmp.setDatabase(dml.getDatabase()); - dmlTmp.setTable(dml.getTable()); - dmlTmp.setType(dml.getType()); - dmlTmp.setTs(dml.getTs()); - dmlTmp.setEs(dml.getEs()); - dmlTmp.setSql(dml.getSql()); - dmlTmp.setData(new ArrayList<>()); - dmlTmp.setOld(new ArrayList<>()); - dmls[i] = dmlTmp; - } - int idx = 0; - for (Map data : dml.getData()) { - int hash; - if (config.getConcurrent()) { - hash = pkHash(config.getDbMapping(), data, threads); - } else { - hash = Math.abs(Math.abs(config.getDbMapping().getTargetTable().hashCode()) % threads); - } - Dml dmlTmp = dmls[hash]; - dmlTmp.getData().add(data); - if (dml.getOld() != null) { - dmlTmp.getOld().add(dml.getOld().get(idx)); - } + public void sync(List simpleDmlList) { + try { + List[] simpleDmlsPartition = simpleDmls2Partition(simpleDmlList); - idx++; - } - for (int i = 0; i < threads; i++) { - Map item = new HashMap<>(); - item.put("dml", dmls[i]); - item.put("config", config); - dmlPartition[i].add(item); - } - - }); List> futures = new ArrayList<>(); for (int i = 0; i < threads; i++) { - int j = i; - futures.add(threadExecutors[i].submit(() -> { - dmlPartition[j].forEach(item -> { - MappingConfig config = (MappingConfig) item.get("config"); - Dml dml = (Dml) item.get("dml"); - sync(config, dml); - }); - batchExecutors[j].commit(); - return true; - })); + if (!simpleDmlsPartition[i].isEmpty()) { + int j = i; + futures.add(threadExecutors[i].submit(() -> { + simpleDmlsPartition[j].forEach(simpleDml -> sync(simpleDml, batchExecutors[j])); + batchExecutors[j].commit(); + return true; + })); + } } - for (int i = 0; i < threads; i++) { + + futures.forEach(future -> { try { - futures.get(i).get(); + future.get(); } catch (Exception e) { // ignore } - } + }); } catch (Exception e) { logger.error("Error rdb sync for batch", e); } } - public void sync(MappingConfig config, Dml dml) { + public void sync(SimpleDml dml, BatchExecutor batchExecutor) { try { - if (config != null) { - String type = dml.getType(); - if (type != null && type.equalsIgnoreCase("INSERT")) { - insert(config, dml); - } else if (type != null && type.equalsIgnoreCase("UPDATE")) { - update(config, dml); - } else if (type != null && type.equalsIgnoreCase("DELETE")) { - delete(config, dml); - } - if (logger.isDebugEnabled()) { - logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); - } + String type = dml.getType(); + if (type != null && type.equalsIgnoreCase("INSERT")) { + insert(dml, batchExecutor); + } else if (type != null && type.equalsIgnoreCase("UPDATE")) { + update(dml, batchExecutor); + } else if (type != null && type.equalsIgnoreCase("DELETE")) { + delete(dml, batchExecutor); } } catch (Exception e) { logger.error(e.getMessage(), e); @@ -162,18 +125,17 @@ public class RdbSyncService { /** * 插入操作 * - * @param config 配置项 - * @param dml DML数据 + * @param simpleDml DML数据 */ - private void insert(MappingConfig config, Dml dml) { - List> data = dml.getData(); + private void insert(SimpleDml simpleDml, BatchExecutor batchExecutor) { + Map data = simpleDml.getData(); if (data == null || data.isEmpty()) { return; } - DbMapping dbMapping = config.getDbMapping(); + DbMapping dbMapping = simpleDml.getConfig().getDbMapping(); - Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data); StringBuilder insertSql = new StringBuilder(); insertSql.append("INSERT INTO ").append(dbMapping.getTargetTable()).append(" ("); @@ -188,185 +150,121 @@ public class RdbSyncService { len = insertSql.length(); insertSql.delete(len - 1, len).append(")"); - Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); + Map ctype = getTargetColumnType(batchExecutor.getConn(), simpleDml.getConfig()); String sql = insertSql.toString(); - Integer tbHash = null; - // 如果不是并行同步的表 - if (!config.getConcurrent()) { - // 按表名hash到一个线程 - tbHash = Math.abs(Math.abs(dbMapping.getTargetTable().hashCode()) % threads); - } - for (Map d : data) { - int hash; - if (tbHash != null) { - hash = tbHash; - } else { - hash = pkHash(dbMapping, d, threads); - } - // ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; - // checkQueue(tpe); - // tpe.submit(() -> { - try { - BatchExecutor batchExecutor = batchExecutors[hash]; - List> values = new ArrayList<>(); - for (Map.Entry entry : columnsMap.entrySet()) { - String targetColumnName = entry.getKey(); - String srcColumnName = entry.getValue(); - if (srcColumnName == null) { - srcColumnName = targetColumnName; - } + try { + List> values = new ArrayList<>(); - Integer type = ctype.get(targetColumnName.toLowerCase()); - if (type == null) { - throw new RuntimeException("No column: " + targetColumnName + " found in target db"); - } - Object value = d.get(srcColumnName); - BatchExecutor.setValue(values, type, value); + for (Map.Entry entry : columnsMap.entrySet()) { + String targetColumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetColumnName; } - batchExecutor.execute(sql, values); - } catch (Exception e) { - logger.error(e.getMessage(), e); + + Integer type = ctype.get(targetColumnName.toLowerCase()); + if (type == null) { + throw new RuntimeException("No column: " + targetColumnName + " found in target db"); + } + Object value = data.get(srcColumnName); + BatchExecutor.setValue(values, type, value); } - // }); + batchExecutor.execute(sql, values); + } catch (Exception e) { + logger.error(e.getMessage(), e); } } /** * 更新操作 * - * @param config 配置项 - * @param dml DML数据 + * @param simpleDml DML数据 */ - private void update(MappingConfig config, Dml dml) { - List> data = dml.getData(); + private void update(SimpleDml simpleDml, BatchExecutor batchExecutor) { + Map data = simpleDml.getData(); if (data == null || data.isEmpty()) { return; } - List> old = dml.getOld(); + Map old = simpleDml.getOld(); if (old == null || old.isEmpty()) { return; } - DbMapping dbMapping = config.getDbMapping(); + DbMapping dbMapping = simpleDml.getConfig().getDbMapping(); - int idx = 1; - Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data.get(0)); + Map columnsMap = SyncUtil.getColumnsMap(dbMapping, data); - Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); + Map ctype = getTargetColumnType(batchExecutor.getConn(), simpleDml.getConfig()); - Integer tbHash = null; - if (!config.getConcurrent()) { - // 按表名hash到一个线程 - tbHash = Math.abs(Math.abs(dbMapping.getTargetTable().hashCode()) % threads); - } - for (Map o : old) { - Map d = data.get(idx - 1); + try { + StringBuilder updateSql = new StringBuilder(); + updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); - int hash; - if (tbHash != null) { - hash = tbHash; - } else { - hash = pkHash(dbMapping, d, o, threads); - } - - // ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; - // checkQueue(tpe); - // tpe.submit(() -> { - try { - BatchExecutor batchExecutor = batchExecutors[hash]; - - StringBuilder updateSql = new StringBuilder(); - updateSql.append("UPDATE ").append(dbMapping.getTargetTable()).append(" SET "); - - List> values = new ArrayList<>(); - for (String srcColumnName : o.keySet()) { - List targetColumnNames = new ArrayList<>(); - columnsMap.forEach((targetColumn, srcColumn) -> { - if (srcColumnName.toLowerCase().equals(srcColumn)) { - targetColumnNames.add(targetColumn); - } - }); - if (!targetColumnNames.isEmpty()) { - for (String targetColumnName : targetColumnNames) { - updateSql.append(targetColumnName).append("=?, "); - Integer type = ctype.get(targetColumnName.toLowerCase()); - BatchExecutor.setValue(values, type, d.get(srcColumnName)); - } + List> values = new ArrayList<>(); + for (String srcColumnName : old.keySet()) { + List targetColumnNames = new ArrayList<>(); + columnsMap.forEach((targetColumn, srcColumn) -> { + if (srcColumnName.toLowerCase().equals(srcColumn)) { + targetColumnNames.add(targetColumn); + } + }); + if (!targetColumnNames.isEmpty()) { + for (String targetColumnName : targetColumnNames) { + updateSql.append(targetColumnName).append("=?, "); + Integer type = ctype.get(targetColumnName.toLowerCase()); + BatchExecutor.setValue(values, type, data.get(srcColumnName)); } } - int len = updateSql.length(); - updateSql.delete(len - 2, len).append(" WHERE "); - - // 拼接主键 - appendCondition(dbMapping, updateSql, ctype, values, d, o); - - batchExecutor.execute(updateSql.toString(), values); - - if (logger.isTraceEnabled()) { - logger.trace("Execute sql: {}", updateSql); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); } - // }); + int len = updateSql.length(); + updateSql.delete(len - 2, len).append(" WHERE "); - idx++; + // 拼接主键 + appendCondition(dbMapping, updateSql, ctype, values, data, old); + + batchExecutor.execute(updateSql.toString(), values); + + if (logger.isTraceEnabled()) { + logger.trace("Execute sql: {}", updateSql); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); } } /** * 删除操作 * - * @param config - * @param dml - * @throws SQLException + * @param simpleDml */ - private void delete(MappingConfig config, Dml dml) { - List> data = dml.getData(); + private void delete(SimpleDml simpleDml, BatchExecutor batchExecutor) { + Map data = simpleDml.getData(); if (data == null || data.isEmpty()) { return; } - DbMapping dbMapping = config.getDbMapping(); + DbMapping dbMapping = simpleDml.getConfig().getDbMapping(); - Map ctype = getTargetColumnType(batchExecutors[0].getConn(), config); - Integer tbHash = null; - if (!config.getConcurrent()) { - // 按表名hash到一个线程 - tbHash = Math.abs(Math.abs(dbMapping.getTargetTable().hashCode()) % threads); - } - for (Map d : data) { - int hash; - if (tbHash != null) { - hash = tbHash; - } else { - hash = pkHash(dbMapping, d, threads); + Map ctype = getTargetColumnType(batchExecutor.getConn(), simpleDml.getConfig()); + + try { + StringBuilder sql = new StringBuilder(); + sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); + + List> values = new ArrayList<>(); + + // 拼接主键 + appendCondition(dbMapping, sql, ctype, values, data); + + batchExecutor.execute(sql.toString(), values); + if (logger.isTraceEnabled()) { + logger.trace("Execute sql: {}", sql); } - - // ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadExecutors[hash]; - // checkQueue(tpe); - // tpe.submit(() -> { - try { - BatchExecutor batchExecutor = batchExecutors[hash]; - StringBuilder sql = new StringBuilder(); - sql.append("DELETE FROM ").append(dbMapping.getTargetTable()).append(" WHERE "); - - List> values = new ArrayList<>(); - - // 拼接主键 - appendCondition(dbMapping, sql, ctype, values, d); - - batchExecutor.execute(sql.toString(), values); - if (logger.isTraceEnabled()) { - logger.trace("Execute sql: {}", sql); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - // }); + } catch (Exception e) { + logger.error(e.getMessage(), e); } } @@ -416,7 +314,8 @@ public class RdbSyncService { */ public static void setPStmt(int type, PreparedStatement pstmt, Object value, int i) throws SQLException { if (value == null) { - pstmt.setObject(i, type); + pstmt.setNull(i, type); + return; } switch (type) { case Types.BIT: @@ -638,18 +537,4 @@ public class RdbSyncService { executorService.shutdown(); } } - - private void checkQueue(ThreadPoolExecutor tpe) { - // 防止队列过大 - while (tpe.getQueue().size() > 10000) { - try { - Thread.sleep(3000); - while (tpe.getQueue().size() > 5000) { - Thread.sleep(1000); - } - } catch (InterruptedException e) { - // ignore - } - } - } } diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java new file mode 100644 index 00000000..cbe45190 --- /dev/null +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java @@ -0,0 +1,102 @@ +package com.alibaba.otter.canal.client.adapter.rdb.support; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; +import com.alibaba.otter.canal.client.adapter.support.Dml; +import org.apache.commons.lang.StringUtils; + +public class SimpleDml { + + private String destination; + private String database; + private String table; + private String type; + private Map data; + private Map old; + + private MappingConfig config; + + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public String getTable() { + return table; + } + + public void setTable(String table) { + this.table = table; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public Map getData() { + return data; + } + + public void setData(Map data) { + this.data = data; + } + + public Map getOld() { + return old; + } + + public void setOld(Map old) { + this.old = old; + } + + public MappingConfig getConfig() { + return config; + } + + public void setConfig(MappingConfig config) { + this.config = config; + } + + public static List dml2SimpleDml(Dml dml, MappingConfig config) { + List simpleDmlList = new ArrayList<>(); + int len = dml.getData().size(); + + String destination = StringUtils.trimToEmpty(dml.getDestination()); + String database = dml.getDatabase(); + String table = dml.getTable(); + + for (int i = 0; i < len; i++) { + SimpleDml simpleDml = new SimpleDml(); + simpleDml.setDestination(dml.getDestination()); + simpleDml.setDatabase(dml.getDatabase()); + simpleDml.setTable(dml.getTable()); + simpleDml.setType(dml.getType()); + simpleDml.setData(dml.getData().get(i)); + if (dml.getOld() != null) { + simpleDml.setOld(dml.getOld().get(i)); + } + simpleDml.setConfig(config); + simpleDmlList.add(simpleDml); + } + + return simpleDmlList; + } +} From 9dc4d78b3b12ae24bec976d239f32bb81f89acfe Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 16 Nov 2018 13:24:36 +0800 Subject: [PATCH 4/4] fix --- .../otter/canal/client/adapter/rdb/support/SimpleDml.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java index cbe45190..d063843a 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SimpleDml.java @@ -6,7 +6,6 @@ import java.util.Map; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; -import org.apache.commons.lang.StringUtils; public class SimpleDml { @@ -79,10 +78,6 @@ public class SimpleDml { List simpleDmlList = new ArrayList<>(); int len = dml.getData().size(); - String destination = StringUtils.trimToEmpty(dml.getDestination()); - String database = dml.getDatabase(); - String table = dml.getTable(); - for (int i = 0; i < len; i++) { SimpleDml simpleDml = new SimpleDml(); simpleDml.setDestination(dml.getDestination());