From f2b8d000f766f644f2332f45daabc84f158aae73 Mon Sep 17 00:00:00 2001 From: wuwo Date: Tue, 6 Nov 2018 16:58:50 +0800 Subject: [PATCH 1/8] skip tsdb snapshot apply if has not receive new ddl change from last snapshot time --- .../canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 8896e2f5..7f4ba057 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 @@ -63,6 +63,7 @@ public class DatabaseTableMeta implements TableMetaTSDB { private CanalEventFilter filter; private CanalEventFilter blackFilter; private EntryPosition lastPosition; + private boolean hasNewDdl; private MetaHistoryDAO metaHistoryDAO; private MetaSnapshotDAO metaSnapshotDAO; private int snapshotInterval = 24; @@ -139,6 +140,7 @@ public class DatabaseTableMeta implements TableMetaTSDB { synchronized (memoryTableMeta) { if (memoryTableMeta.apply(position, schema, ddl, extra)) { this.lastPosition = position; + this.hasNewDdl = true; // 同步每次变更给远程做历史记录 return applyHistoryToDB(position, schema, ddl, extra); } else { @@ -270,10 +272,11 @@ public class DatabaseTableMeta implements TableMetaTSDB { MemoryTableMeta tmpMemoryTableMeta = new MemoryTableMeta(); Map schemaDdls = null; synchronized (memoryTableMeta) { - if (!init && position == null) { + if (!init && !hasNewDdl) { // 如果是持续构建,则识别一下是否有DDL变更过,如果没有就忽略了 return false; } + this.hasNewDdl = false; schemaDdls = memoryTableMeta.snapshot(); for (Map.Entry entry : schemaDdls.entrySet()) { tmpMemoryTableMeta.apply(position, entry.getKey(), entry.getValue(), null); From 87abaee8a7e50ee6a4c7b5f9b3f70ced217e6ce2 Mon Sep 17 00:00:00 2001 From: wuwo Date: Wed, 7 Nov 2018 15:50:33 +0800 Subject: [PATCH 2/8] a little improvement for DatabaseTableMeta --- .../inbound/mysql/tsdb/DatabaseTableMeta.java | 27 ++++++++++++++----- 1 file changed, 20 insertions(+), 7 deletions(-) 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 7f4ba057..a958dbbe 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,8 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.regex.Pattern; import org.apache.commons.beanutils.BeanUtils; @@ -57,6 +59,7 @@ public class DatabaseTableMeta implements TableMetaTSDB { return thread; } }); + private ReadWriteLock lock = new ReentrantReadWriteLock(); private String destination; private MemoryTableMeta memoryTableMeta; private MysqlConnection connection; // 查询meta信息的链接 @@ -129,15 +132,19 @@ public class DatabaseTableMeta implements TableMetaTSDB { @Override public TableMeta find(String schema, String table) { - synchronized (memoryTableMeta) { + lock.readLock().lock(); + try{ return memoryTableMeta.find(schema, table); + }finally { + lock.readLock().unlock(); } } @Override public boolean apply(EntryPosition position, String schema, String ddl, String extra) { // 首先记录到内存结构 - synchronized (memoryTableMeta) { + lock.writeLock().lock(); + try{ if (memoryTableMeta.apply(position, schema, ddl, extra)) { this.lastPosition = position; this.hasNewDdl = true; @@ -146,6 +153,8 @@ public class DatabaseTableMeta implements TableMetaTSDB { } else { throw new RuntimeException("apply to memory is failed"); } + }finally { + lock.writeLock().unlock(); } } @@ -269,18 +278,22 @@ public class DatabaseTableMeta implements TableMetaTSDB { */ private boolean applySnapshotToDB(EntryPosition position, boolean init) { // 获取一份快照 - MemoryTableMeta tmpMemoryTableMeta = new MemoryTableMeta(); Map schemaDdls = null; - synchronized (memoryTableMeta) { + lock.readLock().lock(); + try{ if (!init && !hasNewDdl) { // 如果是持续构建,则识别一下是否有DDL变更过,如果没有就忽略了 return false; } this.hasNewDdl = false; schemaDdls = memoryTableMeta.snapshot(); - for (Map.Entry entry : schemaDdls.entrySet()) { - tmpMemoryTableMeta.apply(position, entry.getKey(), entry.getValue(), null); - } + }finally{ + lock.readLock().unlock(); + } + + MemoryTableMeta tmpMemoryTableMeta = new MemoryTableMeta(); + for (Map.Entry entry : schemaDdls.entrySet()) { + tmpMemoryTableMeta.apply(position, entry.getKey(), entry.getValue(), null); } // 基于临时内存对象进行对比 From 2278340c7b8453e2521bad6bed0b4a82017d47b0 Mon Sep 17 00:00:00 2001 From: littleneko Date: Wed, 7 Nov 2018 20:24:38 +0800 Subject: [PATCH 3/8] =?UTF-8?q?fix:=20buildLastTransactionPosition=20?= =?UTF-8?q?=E7=9A=84=20for=20=E5=BE=AA=E7=8E=AF=E5=B0=91=E4=BA=86=20=3D0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../alibaba/otter/canal/parse/inbound/AbstractEventParser.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index 8902c05e..4d5cf7ec 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -417,7 +417,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle } protected LogPosition buildLastTransactionPosition(List entries) { // 初始化一下 - for (int i = entries.size() - 1; i > 0; i--) { + for (int i = entries.size() - 1; i >= 0; i--) { CanalEntry.Entry entry = entries.get(i); if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {// 尽量记录一个事务做为position return buildLastPosition(entry); From 2c1f3e649aafef87f2ed7255dba513aa34d07e28 Mon Sep 17 00:00:00 2001 From: "jiangahng.loujh" Date: Thu, 8 Nov 2018 13:50:23 +0800 Subject: [PATCH 4/8] fixed issue #1107 , more info message --- .../otter/canal/parse/inbound/AbstractEventParser.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index 4d5cf7ec..cd6e25be 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -181,6 +181,8 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle serverId = queryServerId; } // 4. 获取最后的位置信息 + long start = System.currentTimeMillis(); + logger.warn("---> begin to find start position, it will be long time for reset or first position"); EntryPosition position = findStartPosition(erosaConnection); final EntryPosition startPosition = position; if (startPosition == null) { @@ -191,7 +193,10 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle throw new CanalParseException("can't find init table meta for " + destination + " with position : " + startPosition); } - logger.warn("find start position : {}", startPosition.toString()); + long end = System.currentTimeMillis(); + logger.warn("---> find start position successfully, {}", startPosition.toString() + " cost : " + + (end - start) + + "ms , the next step is binlog dump"); // 重新链接，因为在找position过程中可能有状态，需要断开后重建 erosaConnection.reconnect(); @@ -417,7 +422,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle } protected LogPosition buildLastTransactionPosition(List entries) { // 初始化一下 - for (int i = entries.size() - 1; i >= 0; i--) { + for (int i = entries.size() - 1; i > 0; i--) { CanalEntry.Entry entry = entries.get(i); if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {// 尽量记录一个事务做为position return buildLastPosition(entry); From 0677443ccf9dce72d521aee8eca2113edf5be309 Mon Sep 17 00:00:00 2001 From: "jiangahng.loujh" Date: Thu, 8 Nov 2018 13:57:12 +0800 Subject: [PATCH 5/8] fixed issue #1100 , fix type compare --- .../inbound/mysql/tsdb/DatabaseTableMeta.java | 91 +++++++++++-------- 1 file changed, 55 insertions(+), 36 deletions(-) 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 a958dbbe..ad15a695 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 @@ -47,32 +47,34 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition; */ public class DatabaseTableMeta implements TableMetaTSDB { - public static final EntryPosition INIT_POSITION = new EntryPosition("0", 0L, -2L, -1L); - private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class); - private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); - private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation"); - private static ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { - @Override - public Thread newThread(Runnable r) { - Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]"); - thread.setDaemon(true); - return thread; - } - }); - private ReadWriteLock lock = new ReentrantReadWriteLock(); - private String destination; - private MemoryTableMeta memoryTableMeta; - private MysqlConnection connection; // 查询meta信息的链接 - private CanalEventFilter filter; - private CanalEventFilter blackFilter; - private EntryPosition lastPosition; - private boolean hasNewDdl; - private MetaHistoryDAO metaHistoryDAO; - private MetaSnapshotDAO metaSnapshotDAO; - private int snapshotInterval = 24; - private int snapshotExpire = 360; - private ScheduledFuture scheduleSnapshotFuture; - + public static final EntryPosition INIT_POSITION = new EntryPosition("0", 0L, -2L, -1L); + private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class); + private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); + private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation"); + private static ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { + + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r, + "[scheduler-table-meta-snapshot]"); + thread.setDaemon(true); + return thread; + } + }); + private ReadWriteLock lock = new ReentrantReadWriteLock(); + private String destination; + private MemoryTableMeta memoryTableMeta; + private MysqlConnection connection; // 查询meta信息的链接 + private CanalEventFilter filter; + private CanalEventFilter blackFilter; + private EntryPosition lastPosition; + private boolean hasNewDdl; + private MetaHistoryDAO metaHistoryDAO; + private MetaSnapshotDAO metaSnapshotDAO; + private int snapshotInterval = 24; + private int snapshotExpire = 360; + private ScheduledFuture scheduleSnapshotFuture; + public DatabaseTableMeta(){ } @@ -109,13 +111,13 @@ public class DatabaseTableMeta implements TableMetaTSDB { } return true; } - + @Override public void destory() { if (memoryTableMeta != null) { memoryTableMeta.destory(); } - + if (connection != null) { try { connection.disconnect(); @@ -124,7 +126,7 @@ public class DatabaseTableMeta implements TableMetaTSDB { .getAddress(), e); } } - + if (scheduleSnapshotFuture != null) { scheduleSnapshotFuture.cancel(false); } @@ -133,9 +135,9 @@ public class DatabaseTableMeta implements TableMetaTSDB { @Override public TableMeta find(String schema, String table) { lock.readLock().lock(); - try{ + try { return memoryTableMeta.find(schema, table); - }finally { + } finally { lock.readLock().unlock(); } } @@ -144,7 +146,7 @@ public class DatabaseTableMeta implements TableMetaTSDB { public boolean apply(EntryPosition position, String schema, String ddl, String extra) { // 首先记录到内存结构 lock.writeLock().lock(); - try{ + try { if (memoryTableMeta.apply(position, schema, ddl, extra)) { this.lastPosition = position; this.hasNewDdl = true; @@ -153,7 +155,7 @@ public class DatabaseTableMeta implements TableMetaTSDB { } else { throw new RuntimeException("apply to memory is failed"); } - }finally { + } finally { lock.writeLock().unlock(); } } @@ -280,14 +282,14 @@ public class DatabaseTableMeta implements TableMetaTSDB { // 获取一份快照 Map schemaDdls = null; lock.readLock().lock(); - try{ + try { if (!init && !hasNewDdl) { // 如果是持续构建,则识别一下是否有DDL变更过,如果没有就忽略了 return false; } this.hasNewDdl = false; schemaDdls = memoryTableMeta.snapshot(); - }finally{ + } finally { lock.readLock().unlock(); } @@ -488,7 +490,24 @@ public class DatabaseTableMeta implements TableMetaTSDB { return false; } - if (!StringUtils.equalsIgnoreCase(sourceField.getColumnType(), targetField.getColumnType())) { + // if (!StringUtils.equalsIgnoreCase(sourceField.getColumnType(), + // targetField.getColumnType())) { + // return false; + // } + + // https://github.com/alibaba/canal/issues/1100 + // 支持一下 int vs int(10) + if ((sourceField.isUnsigned() && !targetField.isUnsigned()) + || (!sourceField.isUnsigned() && targetField.isUnsigned())) { + return false; + } + + boolean columnTypeCompare = false; + columnTypeCompare |= StringUtils.containsIgnoreCase(sourceField.getColumnType(), + targetField.getColumnType()); + columnTypeCompare |= StringUtils.containsIgnoreCase(targetField.getColumnType(), + sourceField.getColumnType()); + if (!columnTypeCompare) { return false; } From f9778a0aff910ec187fbd26f52d443547505cc07 Mon Sep 17 00:00:00 2001 From: agapple Date: Fri, 9 Nov 2018 13:58:39 +0800 Subject: [PATCH 6/8] fixed issue #1087, support kafka docker check --- docker/image/admin/app.sh | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/docker/image/admin/app.sh b/docker/image/admin/app.sh index ca9000c2..413378c8 100755 --- a/docker/image/admin/app.sh +++ b/docker/image/admin/app.sh @@ -76,9 +76,9 @@ function checkStart() { function start_canal() { echo "start canal ..." - serverPort=`perl -le 'print $ENV{"canal.port"}'` - if [ -z "$serverPort" ] ; then - serverPort=11111 + metricsPort=`perl -le 'print $ENV{"canal.metrics.pull.port"}'` + if [ -z "$metricsPort" ] ; then + metricsPort=11112 fi destination=`perl -le 'print $ENV{"canal.destinations"}'` @@ -95,7 +95,7 @@ function start_canal() { su admin -c 'cd /home/admin/canal-server/bin/ && sh restart.sh 1>>/tmp/start.log 2>&1' sleep 5 #check start - checkStart "canal" "nc 127.0.0.1 $serverPort -w 1 -z | wc -l" 30 + checkStart "canal" "nc 127.0.0.1 $metricsPort -w 1 -z | wc -l" 30 } function stop_canal() { From 417e04f65584d2758b878249c060b2788f79c7ed Mon Sep 17 00:00:00 2001 From: wuwo Date: Fri, 9 Nov 2018 19:23:03 +0800 Subject: [PATCH 7/8] fixed DatabaseTableMeta#compareTableMeta --- .../parse/inbound/mysql/tsdb/DatabaseTableMeta.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) 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 ad15a695..068db5ed 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 @@ -502,11 +502,13 @@ public class DatabaseTableMeta implements TableMetaTSDB { return false; } + String sign = sourceField.isUnsigned() ? "unsigned" : "signed"; + String sourceColumnType = StringUtils.removeEndIgnoreCase(sourceField.getColumnType(), sign).trim(); + String targetColumnType = StringUtils.removeEndIgnoreCase(targetField.getColumnType(), sign).trim(); + boolean columnTypeCompare = false; - columnTypeCompare |= StringUtils.containsIgnoreCase(sourceField.getColumnType(), - targetField.getColumnType()); - columnTypeCompare |= StringUtils.containsIgnoreCase(targetField.getColumnType(), - sourceField.getColumnType()); + columnTypeCompare |= StringUtils.containsIgnoreCase(sourceColumnType, targetColumnType); + columnTypeCompare |= StringUtils.containsIgnoreCase(targetColumnType, sourceColumnType); if (!columnTypeCompare) { return false; } From cc1574d9138b47b6724d53f38c0bbca15a18715b Mon Sep 17 00:00:00 2001 From: agapple Date: Sun, 11 Nov 2018 01:15:02 +0800 Subject: [PATCH 8/8] fixed adapter --- .../adapter/es/config/ESSyncConfigLoader.java | 8 +++- .../hbase/config/MappingConfigLoader.java | 2 +- .../launcher/src/main/assembly/release.xml | 8 ++-- .../launcher/src/main/bin/startup.bat | 2 +- .../launcher/src/main/bin/startup.sh | 2 +- .../rdb/config/MappingConfigLoader.java | 2 +- .../otter/canal/deployer/CanalLauncher.java | 39 +++++++++++-------- deployer/src/main/resources/canal.properties | 3 +- .../resources/example/instance.properties | 4 +- 9 files changed, 38 insertions(+), 32 deletions(-) diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java index 681537cf..81beede2 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/config/ESSyncConfigLoader.java @@ -5,7 +5,11 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; import java.nio.charset.StandardCharsets; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -99,7 +103,7 @@ public class ESSyncConfigLoader { InputStream in = null; try { // 先取本地文件，再取类路径 - File configFile = new File("../config/" + config); + File configFile = new File("../conf/" + config); if (configFile.exists()) { in = new FileInputStream(configFile); } else { diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java index f2ee6994..2ac3a0d0 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java @@ -125,7 +125,7 @@ public class MappingConfigLoader { InputStream in = null; try { // 先取本地文件，再取类路径 - File configFile = new File("../config/" + config); + File configFile = new File("../conf/" + config); if (configFile.exists()) { in = new FileInputStream(configFile); } else { diff --git a/client-adapter/launcher/src/main/assembly/release.xml b/client-adapter/launcher/src/main/assembly/release.xml index 163f2015..bebad929 100644 --- a/client-adapter/launcher/src/main/assembly/release.xml +++ b/client-adapter/launcher/src/main/assembly/release.xml @@ -24,7 +24,7 @@ ./src/main/resources - /config + /conf **/* @@ -32,21 +32,21 @@ ../elasticsearch/src/main/resources/es - /config/es + /conf/es **/* ../hbase/src/main/resources/hbase - /config/hbase + /conf/hbase **/* ../rdb/src/main/resources/ - /config + /conf META-INF/** diff --git a/client-adapter/launcher/src/main/bin/startup.bat b/client-adapter/launcher/src/main/bin/startup.bat index 00555432..0c464dff 100755 --- a/client-adapter/launcher/src/main/bin/startup.bat +++ b/client-adapter/launcher/src/main/bin/startup.bat @@ -5,7 +5,7 @@ set ENV_PATH=.\ if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0% -set conf_dir=%ENV_PATH%\..\config +set conf_dir=%ENV_PATH%\..\conf set CLASSPATH=%conf_dir% set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH% diff --git a/client-adapter/launcher/src/main/bin/startup.sh b/client-adapter/launcher/src/main/bin/startup.sh index 001f7fb9..63255c59 100644 --- a/client-adapter/launcher/src/main/bin/startup.sh +++ b/client-adapter/launcher/src/main/bin/startup.sh @@ -55,7 +55,7 @@ ADAPTER_OPTS="-DappName=canal-adapter" for i in $base/lib/*; do CLASSPATH=$i:"$CLASSPATH"; done -CLASSPATH="$base/config:$CLASSPATH"; +CLASSPATH="$base/conf:$CLASSPATH"; echo "cd to $bin_abs_path for workaround relative path" cd $bin_abs_path diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java index 8658d45e..aefb467f 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/MappingConfigLoader.java @@ -72,7 +72,7 @@ public class MappingConfigLoader { InputStream in = null; try { // 先取本地文件，再取类路径 - File configFile = new File("../config/" + config); + File configFile = new File("../conf/" + config); if (configFile.exists()) { in = new FileInputStream(configFile); } else { 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 d1663305..4b85f50c 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 @@ -3,11 +3,11 @@ package com.alibaba.otter.canal.deployer; import java.io.FileInputStream; import java.util.Properties; -import com.alibaba.otter.canal.common.MQProperties; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.kafka.CanalKafkaProducer; import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer; import com.alibaba.otter.canal.server.CanalMQStarter; @@ -73,22 +73,7 @@ public class CanalLauncher { if (canalMQProducer != null) { CanalMQStarter canalMQStarter = new CanalMQStarter(canalMQProducer); - MQProperties mqProperties = new MQProperties(); - mqProperties.setServers(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS)); - mqProperties - .setRetries(Integer.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_RETRIES))); - mqProperties - .setBatchSize(Integer.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BATCHSIZE))); - mqProperties - .setLingerMs(Integer.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_LINGERMS))); - mqProperties.setBufferMemory( - Long.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BUFFERMEMORY))); - mqProperties.setCanalBatchSize( - Integer.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALBATCHSIZE))); - mqProperties.setCanalGetTimeout( - Long.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_CANALGETTIMEOUT))); - mqProperties.setFlatMessage( - Boolean.valueOf(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_FLATMESSAGE))); + MQProperties mqProperties = buildMQPosition(properties); canalMQStarter.start(mqProperties); controller.setCanalMQStarter(canalMQStarter); } @@ -98,6 +83,26 @@ public class CanalLauncher { } } + private static MQProperties buildMQPosition(Properties properties) { + MQProperties mqProperties = new MQProperties(); + mqProperties.setServers(CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS)); + mqProperties.setRetries(Integer.valueOf(CanalController.getProperty(properties, + CanalConstants.CANAL_MQ_RETRIES))); + mqProperties.setBatchSize(Integer.valueOf(CanalController.getProperty(properties, + CanalConstants.CANAL_MQ_BATCHSIZE))); + mqProperties.setLingerMs(Integer.valueOf(CanalController.getProperty(properties, + CanalConstants.CANAL_MQ_LINGERMS))); + mqProperties.setBufferMemory(Long.valueOf(CanalController.getProperty(properties, + CanalConstants.CANAL_MQ_BUFFERMEMORY))); + mqProperties.setCanalBatchSize(Integer.valueOf(CanalController.getProperty(properties, + CanalConstants.CANAL_MQ_CANALBATCHSIZE))); + mqProperties.setCanalGetTimeout(Long.valueOf(CanalController.getProperty(properties, + CanalConstants.CANAL_MQ_CANALGETTIMEOUT))); + mqProperties.setFlatMessage(Boolean.valueOf(CanalController.getProperty(properties, + CanalConstants.CANAL_MQ_FLATMESSAGE))); + return mqProperties; + } + private static void setGlobalUncaughtExceptionHandler() { Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 0ff5997a..cf2a225e 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -101,12 +101,11 @@ canal.instance.global.spring.xml = classpath:spring/file-instance.xml ################################################## ######### MQ ############# ################################################## -canal.mq.servers=slave1:6667 +canal.mq.servers=127.0.0.1:6667 canal.mq.retries=0 canal.mq.batchSize=16384 canal.mq.lingerMs=1 canal.mq.bufferMemory=33554432 - canal.mq.canalBatchSize=50 canal.mq.canalGetTimeout=100 canal.mq.flatMessage=true \ No newline at end of file diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index ee3ed601..3a9ac883 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -45,10 +45,8 @@ canal.instance.filter.black.regex= # mq config canal.mq.topic=example -# 单分区指定分区0 canal.mq.partition=0 -# 多分区指定分区数量 +# hash partition config #canal.mq.partitionsNum=3 -# 多分区指定库表主键, 规则: 库名.表名:唯一主键, 多个按逗号分隔 #canal.mq.partitionHash=mytest.person:id,mytest.role:id ################################################# \ No newline at end of file