chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed adapter

Browse Source
This commit is contained in:
agapple
2018-11-11 01:15:02 +08:00
parent 808bedf984
commit cc1574d913
9 changed files with 38 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
View File

@@ -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() {

3
deployer/src/main/resources/canal.properties
View File

@@ -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

4
deployer/src/main/resources/example/instance.properties
View File

@@ -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
#################################################