From 8af1301062ae1b9402ed2c26279027513de2d3d7 Mon Sep 17 00:00:00 2001 From: Renkai Date: Tue, 13 Nov 2018 18:29:25 +0800 Subject: [PATCH 1/3] =?UTF-8?q?Kafka=E5=A2=9E=E5=8A=A0=E5=8F=82=E6=95=B0ac?= =?UTF-8?q?ks=E5=92=8CcompressionType?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/deployer/CanalConstants.java | 2 ++ .../otter/canal/deployer/CanalLauncher.java | 2 ++ .../otter/canal/common/MQProperties.java | 19 +++++++++++++++++++ .../otter/canal/kafka/CanalKafkaProducer.java | 3 ++- 4 files changed, 25 insertions(+), 1 deletion(-) diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index 2bfb32ea..aa9b2909 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -45,6 +45,8 @@ public class CanalConstants { public static final String CANAL_MQ_CANALBATCHSIZE = ROOT + "." + "mq.canalBatchSize"; public static final String CANAL_MQ_CANALGETTIMEOUT = ROOT + "." + "mq.canalGetTimeout"; public static final String CANAL_MQ_FLATMESSAGE = ROOT + "." + "mq.flatMessage"; + public static final String CANAL_MQ_COMPRESSION_TYPE = ROOT + "." + "mq.compressionType"; + public static final String CANAL_MQ_ACKS = ROOT + "." + "mq.acks"; public static String getInstanceModeKey(String destination) { return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 4b85f50c..f64e6184 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -100,6 +100,8 @@ public class CanalLauncher { CanalConstants.CANAL_MQ_CANALGETTIMEOUT))); mqProperties.setFlatMessage(Boolean.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_FLATMESSAGE))); + mqProperties.setCompressionType(CanalController.getProperty(properties,CanalConstants.CANAL_MQ_COMPRESSION_TYPE)); + mqProperties.setAcks(CanalController.getProperty(properties,CanalConstants.CANAL_MQ_ACKS)); return mqProperties; } diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index 29375e2f..3bc3394b 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -20,6 +20,8 @@ public class MQProperties { private int canalBatchSize = 50; private Long canalGetTimeout = 100L; private boolean flatMessage = true; + private String compressionType = "none"; + private String acks = "all"; public static class CanalDestination { @@ -149,4 +151,21 @@ public class MQProperties { public void setProducerGroup(String producerGroup) { this.producerGroup = producerGroup; } + + public String getCompressionType() { + return compressionType; + } + + public void setCompressionType(String compressionType) { + this.compressionType = compressionType; + } + + public String getAcks() { + return acks; + } + + public void setAcks(String acks) { + this.acks = acks; + } + } diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 78a67c13..d7cf05bb 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -37,7 +37,8 @@ public class CanalKafkaProducer implements CanalMQProducer { this.kafkaProperties = kafkaProperties; Properties properties = new Properties(); properties.put("bootstrap.servers", kafkaProperties.getServers()); - properties.put("acks", "all"); + properties.put("acks", kafkaProperties.getAcks()); + properties.put("compression.type",kafkaProperties.getCompressionType()); properties.put("retries", kafkaProperties.getRetries()); properties.put("batch.size", kafkaProperties.getBatchSize()); properties.put("linger.ms", kafkaProperties.getLingerMs()); From 20062d861d2236ffa381eecd430ff789856c0e65 Mon Sep 17 00:00:00 2001 From: agapple Date: Tue, 13 Nov 2018 19:59:18 +0800 Subject: [PATCH 2/3] fixed test code --- .../canal/client/adapter/support/ExtensionLoader.java | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java index b29bff4a..860cb82b 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java @@ -2,13 +2,15 @@ package com.alibaba.otter.canal.client.adapter.support; import java.io.BufferedReader; import java.io.File; -import java.io.IOException; import java.io.InputStreamReader; import java.net.MalformedURLException; import java.net.URL; import java.net.URLClassLoader; import java.nio.file.Paths; -import java.util.*; +import java.util.Arrays; +import java.util.Enumeration; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.regex.Pattern; @@ -178,8 +180,9 @@ public class ExtensionLoader { @SuppressWarnings("unchecked") private T createExtension(String name, String key) { - System.out.println("xxxxxxxxxxxxx"); - getExtensionClasses().forEach((k, v) -> logger.info("fffff: " + k + " " + v.getName())); + // System.out.println("xxxxxxxxxxxxx"); + // getExtensionClasses().forEach((k, v) -> logger.info("fffff: " + k + + // " " + v.getName())); Class clazz = getExtensionClasses().get(name); if (clazz == null) { throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type From 5ad9cf3f163d73ceaf84f66959978f0e6d5c044f Mon Sep 17 00:00:00 2001 From: wuwo Date: Tue, 13 Nov 2018 20:54:36 +0800 Subject: [PATCH 3/3] let tsdbSnapshotInterval config work correctly --- .../manager/CanalInstanceWithManager.java | 2 +- .../inbound/mysql/LocalBinlogEventParser.java | 1 + .../parse/inbound/mysql/MysqlEventParser.java | 1 + .../inbound/mysql/tsdb/DatabaseTableMeta.java | 54 ++++++++++--------- .../mysql/tsdb/TableMetaTSDBBuilder.java | 1 - 5 files changed, 32 insertions(+), 27 deletions(-) diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index 154cd53f..11091695 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -304,7 +304,6 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { } boolean tsdbEnable = BooleanUtils.toBoolean(parameters.getTsdbEnable()); if (tsdbEnable) { - mysqlEventParser.setEnableTsdb(tsdbEnable); mysqlEventParser.setTableMetaTSDBFactory(new DefaultTableMetaTSDBFactory() { @Override @@ -327,6 +326,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { } } }); + mysqlEventParser.setEnableTsdb(tsdbEnable); } eventParser = mysqlEventParser; } else if (type.isLocalBinlog()) { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java index 4f6ace4f..db235d9b 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java @@ -57,6 +57,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements ((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter); ((DatabaseTableMeta) tableMetaTSDB).setSnapshotInterval(tsdbSnapshotInterval); ((DatabaseTableMeta) tableMetaTSDB).setSnapshotExpire(tsdbSnapshotExpire); + ((DatabaseTableMeta) tableMetaTSDB).init(destination); } tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index e8ca45e3..167d2460 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -124,6 +124,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE ((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter); ((DatabaseTableMeta) tableMetaTSDB).setSnapshotInterval(tsdbSnapshotInterval); ((DatabaseTableMeta) tableMetaTSDB).setSnapshotExpire(tsdbSnapshotExpire); + ((DatabaseTableMeta) tableMetaTSDB).init(destination); } tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index 068db5ed..dd90ffbf 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -10,6 +10,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.regex.Pattern; @@ -62,9 +63,10 @@ public class DatabaseTableMeta implements TableMetaTSDB { } }); private ReadWriteLock lock = new ReentrantReadWriteLock(); + private AtomicBoolean initialized = new AtomicBoolean(false); private String destination; private MemoryTableMeta memoryTableMeta; - private MysqlConnection connection; // 查询meta信息的链接 + private volatile MysqlConnection connection; // 查询meta信息的链接 private CanalEventFilter filter; private CanalEventFilter blackFilter; private EntryPosition lastPosition; @@ -74,40 +76,42 @@ public class DatabaseTableMeta implements TableMetaTSDB { private int snapshotInterval = 24; private int snapshotExpire = 360; private ScheduledFuture scheduleSnapshotFuture; - + public DatabaseTableMeta(){ } @Override public boolean init(final String destination) { - this.destination = destination; - this.memoryTableMeta = new MemoryTableMeta(); + if (initialized.compareAndSet(false, true)) { + this.destination = destination; + this.memoryTableMeta = new MemoryTableMeta(); - // 24小时生成一份snapshot - if (snapshotInterval > 0) { - scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(new Runnable() { + // 24小时生成一份snapshot + if (snapshotInterval > 0) { + scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(new Runnable() { - @Override - public void run() { - boolean applyResult = false; - try { - MDC.put("destination", destination); - applyResult = applySnapshotToDB(lastPosition, false); - } catch (Throwable e) { - logger.error("scheudle applySnapshotToDB faield", e); - } - - try { - MDC.put("destination", destination); - if (applyResult) { - snapshotExpire((int) TimeUnit.HOURS.toSeconds(snapshotExpire)); + @Override + public void run() { + boolean applyResult = false; + try { + MDC.put("destination", destination); + applyResult = applySnapshotToDB(lastPosition, false); + } catch (Throwable e) { + logger.error("scheudle applySnapshotToDB faield", e); + } + + try { + MDC.put("destination", destination); + if (applyResult) { + snapshotExpire((int) TimeUnit.HOURS.toSeconds(snapshotExpire)); + } + } catch (Throwable e) { + logger.error("scheudle snapshotExpire faield", e); } - } catch (Throwable e) { - logger.error("scheudle snapshotExpire faield", e); } - } - }, snapshotInterval, snapshotInterval, TimeUnit.HOURS); + }, snapshotInterval, snapshotInterval, TimeUnit.HOURS); + } } return true; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java index 107e1013..75b97f14 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java @@ -36,7 +36,6 @@ public class TableMetaTSDBBuilder { } } TableMetaTSDB tableMetaTSDB = (TableMetaTSDB) applicationContext.getBean("tableMetaTSDB"); - tableMetaTSDB.init(destination); logger.info("{} init TableMetaTSDB with {}", destination, springXml); return tableMetaTSDB; } else {