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 a2a2d532..2bfb32ea 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 @@ -37,6 +37,15 @@ public class CanalConstants { public static final String CANAL_SOCKETCHANNEL = ROOT + "." + "socketChannel"; + public static final String CANAL_MQ_SERVERS = ROOT + "." + "mq.servers"; + public static final String CANAL_MQ_RETRIES = ROOT + "." + "mq.retries"; + public static final String CANAL_MQ_BATCHSIZE = ROOT + "." + "mq.batchSize"; + public static final String CANAL_MQ_LINGERMS = ROOT + "." + "mq.lingerMs"; + public static final String CANAL_MQ_BUFFERMEMORY = ROOT + "." + "mq.bufferMemory"; + 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 String getInstanceModeKey(String destination) { return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index a15921de..535dbd44 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -17,6 +17,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import com.alibaba.otter.canal.common.CanalException; +import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.common.zookeeper.ZkClientx; import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils; @@ -35,6 +36,7 @@ import com.alibaba.otter.canal.instance.manager.CanalConfigClient; import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator; import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator; import com.alibaba.otter.canal.parse.CanalEventParser; +import com.alibaba.otter.canal.server.CanalMQStarter; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; import com.alibaba.otter.canal.server.exception.CanalServerException; import com.alibaba.otter.canal.server.netty.CanalServerWithNetty; @@ -68,6 +70,8 @@ public class CanalController { private CanalInstanceGenerator instanceGenerator; private ZkClientx zkclientx; + private CanalMQStarter canalMQStarter; + public CanalController(){ this(System.getProperties()); } @@ -105,7 +109,7 @@ public class CanalController { logger.info("No valid metrics server port found, use default 11112."); embededCanalServer.setMetricsPort(11112); } - + String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY); if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) { canalServer = CanalServerWithNetty.instance(); @@ -127,80 +131,81 @@ public class CanalController { final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port); ServerRunningMonitors.setServerData(serverData); - ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function() { + ServerRunningMonitors + .setRunningMonitors(MigrateMap.makeComputingMap(new Function() { - public ServerRunningMonitor apply(final String destination) { - ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData); - runningMonitor.setDestination(destination); - runningMonitor.setListener(new ServerRunningListener() { + public ServerRunningMonitor apply(final String destination) { + ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData); + runningMonitor.setDestination(destination); + runningMonitor.setListener(new ServerRunningListener() { - public void processActiveEnter() { - try { - MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); - embededCanalServer.start(destination); - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); - } - } - - public void processActiveExit() { - try { - MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); - embededCanalServer.stop(destination); - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); - } - } - - public void processStart() { - try { - if (zkclientx != null) { - final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" - + port); - initCid(path); - zkclientx.subscribeStateChanges(new IZkStateListener() { - - public void handleStateChanged(KeeperState state) throws Exception { - - } - - public void handleNewSession() throws Exception { - initCid(path); - } - - @Override - public void handleSessionEstablishmentError(Throwable error) throws Exception { - logger.error("failed to connect to zookeeper", error); - } - }); + public void processActiveEnter() { + try { + MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); + embededCanalServer.start(destination); + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); } - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); } - } - public void processStop() { - try { - MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); - if (zkclientx != null) { - final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" - + port); - releaseCid(path); + public void processActiveExit() { + try { + MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); + embededCanalServer.stop(destination); + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); } - } finally { - MDC.remove(CanalConstants.MDC_DESTINATION); } - } - }); - if (zkclientx != null) { - runningMonitor.setZkClient(zkclientx); + public void processStart() { + try { + if (zkclientx != null) { + final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, + ip + ":" + port); + initCid(path); + zkclientx.subscribeStateChanges(new IZkStateListener() { + + public void handleStateChanged(KeeperState state) throws Exception { + + } + + public void handleNewSession() throws Exception { + initCid(path); + } + + @Override + public void handleSessionEstablishmentError(Throwable error) throws Exception { + logger.error("failed to connect to zookeeper", error); + } + }); + } + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); + } + } + + public void processStop() { + try { + MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); + if (zkclientx != null) { + final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, + ip + ":" + port); + releaseCid(path); + } + } finally { + MDC.remove(CanalConstants.MDC_DESTINATION); + } + } + + }); + if (zkclientx != null) { + runningMonitor.setZkClient(zkclientx); + } + // 触发创建一下cid节点 + runningMonitor.init(); + return runningMonitor; } - // 触发创建一下cid节点 - runningMonitor.init(); - return runningMonitor; - } - })); + })); // 初始化monitor机制 autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN)); @@ -220,6 +225,9 @@ public class CanalController { ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination); if (!config.getLazy() && !runningMonitor.isStart()) { runningMonitor.start(); + if (canalMQStarter != null) { + canalMQStarter.start(getMQProperties(properties)); + } } } } @@ -228,6 +236,9 @@ public class CanalController { // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息 InstanceConfig config = instanceConfigs.remove(destination); if (config != null) { + if (canalMQStarter != null) { + canalMQStarter.stop(); + } embededCanalServer.stop(destination); ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination); if (runningMonitor.isStart()) { @@ -246,7 +257,8 @@ public class CanalController { instanceConfigMonitors = MigrateMap.makeComputingMap(new Function() { public InstanceConfigMonitor apply(InstanceMode mode) { - int scanInterval = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL)); + int scanInterval = Integer + .valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL)); if (mode.isSpring()) { SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor(); @@ -354,8 +366,8 @@ public class CanalController { InstanceConfig oldConfig = instanceConfigs.put(destination, config); if (oldConfig != null) { - logger.warn("destination:{} old config:{} has replace by new config:{}", new Object[] { destination, - oldConfig, config }); + logger + .warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config); } } } @@ -514,4 +526,31 @@ public class CanalController { } } + public CanalMQStarter getCanalMQStarter() { + return canalMQStarter; + } + + public void setCanalMQStarter(CanalMQStarter canalMQStarter) { + this.canalMQStarter = canalMQStarter; + } + + public static MQProperties getMQProperties(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; + } } 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 3277c8c1..8ad3d06b 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 @@ -71,10 +71,9 @@ public class CanalLauncher { }); if (canalMQProducer != null) { - CanalMQStarter canalServerStarter = new CanalMQStarter(canalMQProducer); - if (canalServerStarter != null) { - canalServerStarter.init(); - } + CanalMQStarter canalMQStarter = new CanalMQStarter(canalMQProducer); + canalMQStarter.start(CanalController.getMQProperties(properties)); + controller.setCanalMQStarter(canalMQStarter); } } catch (Throwable e) { logger.error("## Something goes wrong when starting up the canal Server:", e); diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index b5284dec..0ff5997a 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -91,9 +91,22 @@ canal.auto.scan.interval = 5 canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml #canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml -canal.instance.global.mode = spring +canal.instance.global.mode = spring canal.instance.global.lazy = false #canal.instance.global.manager.address = 127.0.0.1:1099 #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml canal.instance.global.spring.xml = classpath:spring/file-instance.xml #canal.instance.global.spring.xml = classpath:spring/default-instance.xml + +################################################## +######### MQ ############# +################################################## +canal.mq.servers=slave1: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 0caef6d2..c2e42618 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -18,7 +18,7 @@ canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info -canal.instance.tsdb.enable=true +canal.instance.tsdb.enable=false #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal @@ -30,8 +30,8 @@ canal.instance.tsdb.enable=true #canal.instance.standby.gtid= # username/password -canal.instance.dbUsername=canal -canal.instance.dbPassword=canal +canal.instance.dbUsername=root +canal.instance.dbPassword=121212 canal.instance.connectionCharset = UTF-8 canal.instance.defaultDatabaseName =test # enable druid Decrypt database password @@ -42,4 +42,13 @@ canal.instance.enableDruid=false canal.instance.filter.regex=.*\\..* # table black regex canal.instance.filter.black.regex= + +# mq config +canal.mq.topic=example +# 但分区指定分区0 +canal.mq.partition=0 +# 多分区指定分区数量 +#canal.mq.partitionsNum=3 +# 多分区指定库表主键, 规则: 库名.表名:唯一主键, 多个按逗号分隔 +#canal.mq.partitionHash=mytest.person:id,mytest.role:id ################################################# \ No newline at end of file diff --git a/deployer/src/main/resources/mq.yml b/deployer/src/main/resources/mq.yml deleted file mode 100644 index 1797bd9b..00000000 --- a/deployer/src/main/resources/mq.yml +++ /dev/null @@ -1,23 +0,0 @@ -servers: slave1:6667 #for rocketmq: means the nameserver -retries: 0 -batchSize: 16384 -lingerMs: 1 -bufferMemory: 33554432 - -# Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) -canalBatchSize: 50 -# Canal get数据的超时时间, 单位: 毫秒, 0为不限超时 -canalGetTimeout: 100 -flatMessage: true - -canalDestinations: - - canalDestination: example - topic: example - partition: -# #对应topic分区数量 -# partitionsNum: 3 -# partitionHash: -# #库名.表名: 唯一主键 -# mytest.person: id - - diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index f8c19bd0..f312b1ce 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -29,6 +29,9 @@ + + + @@ -187,4 +190,11 @@ + + + + + + + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index 5524f73f..3b6d6a8d 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -29,6 +29,9 @@ + + + @@ -172,4 +175,11 @@ + + + + + + + diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index e614b86c..d9c61fb8 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -29,6 +29,9 @@ + + + @@ -262,4 +265,11 @@ + + + + + + + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index 8fdfb178..9260c166 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -29,6 +29,9 @@ + + + @@ -160,4 +163,11 @@ + + + + + + + diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java index 9143a626..b1b64d5b 100644 --- a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/AbstractCanalInstance.java @@ -38,6 +38,9 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can protected CanalEventSink> eventSink; // 链接parse和store的桥接器 protected CanalMetaManager metaManager; // 消费信息管理器 protected CanalAlarmHandler alarmHandler; // alarm报警机制 + protected CanalMQConfig mqConfig; // mq的配置 + + @Override public boolean subscribeChange(ClientIdentity identity) { @@ -241,4 +244,9 @@ public class AbstractCanalInstance extends AbstractCanalLifeCycle implements Can public CanalAlarmHandler getAlarmHandler() { return alarmHandler; } + + @Override + public CanalMQConfig getMqConfig() { + return mqConfig; + } } diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java index 3bd0464b..3d1a08c2 100644 --- a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java @@ -32,4 +32,6 @@ public interface CanalInstance extends CanalLifeCycle { * 客户端发生订阅/取消订阅行为 */ boolean subscribeChange(ClientIdentity identity); + + CanalMQConfig getMqConfig(); } diff --git a/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java new file mode 100644 index 00000000..c275f150 --- /dev/null +++ b/instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalMQConfig.java @@ -0,0 +1,68 @@ +package com.alibaba.otter.canal.instance.core; + +import java.util.LinkedHashMap; +import java.util.Map; + +public class CanalMQConfig { + + private String topic; + private Integer partition; + private Integer partitionsNum; + private String partitionHash; + + private volatile Map partitionHashProperties; + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + public Integer getPartition() { + return partition; + } + + public void setPartition(Integer partition) { + this.partition = partition; + } + + public Integer getPartitionsNum() { + return partitionsNum; + } + + public void setPartitionsNum(Integer partitionsNum) { + this.partitionsNum = partitionsNum; + } + + public String getPartitionHash() { + return partitionHash; + } + + public void setPartitionHash(String partitionHash) { + this.partitionHash = partitionHash; + } + + public Map getPartitionHashProperties() { + if (partitionHashProperties == null) { + synchronized (CanalMQConfig.class) { + if (partitionHashProperties == null) { + if (partitionHash != null) { + partitionHashProperties = new LinkedHashMap<>(); + String[] items = partitionHash.split(","); + for (String item : items) { + int i = item.indexOf(":"); + if (i > -1) { + String dbTable = item.substring(0, i).trim(); + String pk = item.substring(i + 1).trim(); + partitionHashProperties.put(dbTable, pk); + } + } + } + } + } + } + return partitionHashProperties; + } +} diff --git a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java index 79693c4c..69672a6c 100644 --- a/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java +++ b/instance/spring/src/main/java/com/alibaba/otter/canal/instance/spring/CanalInstanceWithSpring.java @@ -2,6 +2,8 @@ package com.alibaba.otter.canal.instance.spring; import java.util.List; +import com.alibaba.otter.canal.instance.core.CanalInstance; +import com.alibaba.otter.canal.instance.core.CanalMQConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -56,4 +58,8 @@ public class CanalInstanceWithSpring extends AbstractCanalInstance { this.alarmHandler = alarmHandler; } + public void setMqConfig(CanalMQConfig mqConfig){ + this.mqConfig = mqConfig; + } + } diff --git a/server/pom.xml b/server/pom.xml index cc4b63d0..3d3084e7 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -26,11 +26,6 @@ ${project.version} - - org.yaml - snakeyaml - 1.17 - org.apache.kafka kafka_2.11 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 14db1deb..c61710e1 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 @@ -1,7 +1,5 @@ package com.alibaba.otter.canal.common; -import java.util.ArrayList; -import java.util.List; import java.util.Map; /** @@ -12,18 +10,19 @@ import java.util.Map; */ public class MQProperties { - private String servers = "127.0.0.1:6667"; - private int retries = 0; - private int batchSize = 16384; - private int lingerMs = 1; - private long bufferMemory = 33554432L; - private boolean filterTransactionEntry = true; - private String producerGroup = "Canal-Producer"; - private int canalBatchSize = 50; - private Long canalGetTimeout = 100L; - private boolean flatMessage = true; + private String servers = "127.0.0.1:6667"; + private int retries = 0; + private int batchSize = 16384; + private int lingerMs = 1; + private long bufferMemory = 33554432L; + private boolean filterTransactionEntry = true; + private String producerGroup = "Canal-Producer"; + private int canalBatchSize = 50; + private Long canalGetTimeout = 100L; + private boolean flatMessage = true; - private List canalDestinations = new ArrayList(); + // private List canalDestinations = new + // ArrayList(); public static class CanalDestination { @@ -138,14 +137,6 @@ public class MQProperties { this.flatMessage = flatMessage; } - public List getCanalDestinations() { - return canalDestinations; - } - - public void setCanalDestinations(List canalDestinations) { - this.canalDestinations = canalDestinations; - } - public boolean isFilterTransactionEntry() { return filterTransactionEntry; } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index 2e861d3f..32cfb762 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -1,17 +1,15 @@ package com.alibaba.otter.canal.server; -import java.io.FileInputStream; -import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; -import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.yaml.snakeyaml.Yaml; import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.instance.core.CanalInstance; +import com.alibaba.otter.canal.instance.core.CanalMQConfig; import com.alibaba.otter.canal.kafka.CanalKafkaProducer; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; @@ -20,37 +18,28 @@ import com.alibaba.otter.canal.spi.CanalMQProducer; public class CanalMQStarter { - private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class); + private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class); - private static final String CLASSPATH_URL_PREFIX = "classpath:"; + private volatile boolean running = false; - private volatile boolean running = false; + private ExecutorService executorService; - private ExecutorService executorService; + private CanalMQProducer canalMQProducer; - private CanalMQProducer canalMQProducer; + private MQProperties properties; - private MQProperties properties; + private CanalServerWithEmbedded canalServer; public CanalMQStarter(CanalMQProducer canalMQProducer){ this.canalMQProducer = canalMQProducer; } - public void init() { + public synchronized void start(MQProperties properties) { try { - logger.info("## load MQ configurations"); - String conf = System.getProperty("mq.conf", "classpath:mq.yml"); - - if (conf.startsWith(CLASSPATH_URL_PREFIX)) { - conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX); - properties = new Yaml().loadAs(CanalMQStarter.class.getClassLoader().getResourceAsStream(conf), - MQProperties.class); - } else { - properties = new Yaml().loadAs(new FileInputStream(conf), MQProperties.class); + if (running) { + return; } - - // 初始化 kafka producer - // canalMQProducer = new CanalKafkaProducer(); + this.properties = properties; canalMQProducer.init(properties); // set filterTransactionEntry if (properties.isFilterTransactionEntry()) { @@ -62,36 +51,34 @@ public class CanalMQStarter { System.setProperty("canal.instance.memory.rawEntry", "false"); } + canalServer = CanalServerWithEmbedded.instance(); + // 对应每个instance启动一个worker线程 - List destinations = properties.getCanalDestinations(); - - executorService = Executors.newFixedThreadPool(destinations.size()); - + executorService = Executors.newFixedThreadPool(canalServer.getCanalInstances().size()); logger.info("## start the MQ workers."); - for (final MQProperties.CanalDestination destination : destinations) { + for (final CanalInstance canalInstance : canalServer.getCanalInstances().values()) { executorService.execute(new Runnable() { @Override public void run() { + MQProperties.CanalDestination destination = new MQProperties.CanalDestination(); + destination.setCanalDestination(canalInstance.getDestination()); + CanalMQConfig mqConfig = canalInstance.getMqConfig(); + destination.setTopic(mqConfig.getTopic()); + destination.setPartition(mqConfig.getPartition()); + destination.setPartitionsNum(mqConfig.getPartitionsNum()); + destination.setPartitionHash(mqConfig.getPartitionHashProperties()); worker(destination); } }); } + running = true; logger.info("## the MQ workers is running now ......"); Runtime.getRuntime().addShutdownHook(new Thread() { public void run() { - try { - logger.info("## stop the MQ workers"); - running = false; - executorService.shutdown(); - canalMQProducer.stop(); - } catch (Throwable e) { - logger.warn("##something goes wrong when stopping MQ workers:", e); - } finally { - logger.info("## canal MQ is down."); - } + stop(); } }); @@ -102,15 +89,31 @@ public class CanalMQStarter { } } + public synchronized void stop() { + if (!running) { + return; + } + try { + logger.info("## stop the MQ workers"); + running = false; + executorService.shutdown(); + canalMQProducer.stop(); + } catch (Throwable e) { + logger.warn("##something goes wrong when stopping MQ workers:", e); + } finally { + logger.info("## canal MQ is down."); + } + } + private void worker(MQProperties.CanalDestination destination) { while (!running) ; - logger.info("## start the canal consumer: {}.", destination.getCanalDestination()); - final CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); + logger.info("## start the MQ producer: {}.", destination.getCanalDestination()); + final ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); while (running) { try { - if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { + if (!canalServer.getCanalInstances().containsKey(clientIdentity.getDestination())) { try { Thread.sleep(3000); } catch (InterruptedException e) { @@ -118,17 +121,18 @@ public class CanalMQStarter { } continue; } - server.subscribe(clientIdentity); - logger.info("## the canal consumer {} is running now ......", destination.getCanalDestination()); + canalServer.subscribe(clientIdentity); + logger.info("## the MQ producer: {} is running now ......", destination.getCanalDestination()); Long getTimeout = properties.getCanalGetTimeout(); int getBatchSize = properties.getCanalBatchSize(); while (running) { Message message; if (getTimeout != null && getTimeout > 0) { - message = server.getWithoutAck(clientIdentity, getBatchSize, getTimeout, TimeUnit.MILLISECONDS); + message = canalServer + .getWithoutAck(clientIdentity, getBatchSize, getTimeout, TimeUnit.MILLISECONDS); } else { - message = server.getWithoutAck(clientIdentity, getBatchSize); + message = canalServer.getWithoutAck(clientIdentity, getBatchSize); } final long batchId = message.getId(); @@ -139,12 +143,12 @@ public class CanalMQStarter { @Override public void commit() { - server.ack(clientIdentity, batchId); // 提交确认 + canalServer.ack(clientIdentity, batchId); // 提交确认 } @Override public void rollback() { - server.rollback(clientIdentity, batchId); + canalServer.rollback(clientIdentity, batchId); } }); // 发送message到topic } else {