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