From 339ecdd81729f17f30d15538125ce5aa185e532f Mon Sep 17 00:00:00 2001 From: mcy Date: Thu, 29 Nov 2018 17:38:13 +0800 Subject: [PATCH] fixed #1229 --- .../otter/canal/deployer/CanalLauncher.java | 2 + .../otter/canal/server/CanalMQStarter.java | 46 +++++++++++-------- 2 files changed, 28 insertions(+), 20 deletions(-) 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 aa7f53b0..fd52f3fa 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 @@ -50,6 +50,8 @@ public class CanalLauncher { if (canalMQProducer != null) { // disable netty System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true"); + System.setProperty(CanalConstants.CANAL_DESTINATIONS, + properties.getProperty(CanalConstants.CANAL_DESTINATIONS)); } logger.info("## start the canal server."); 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 ade0adaf..a283e23f 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 @@ -7,6 +7,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,9 +61,12 @@ public class CanalMQStarter { // 对应每个instance启动一个worker线程 executorService = Executors.newCachedThreadPool(); logger.info("## start the MQ workers."); - for (final CanalInstance canalInstance : canalServer.getCanalInstances().values()) { - CanalMQRunnable canalMQRunnable = new CanalMQRunnable(canalInstance); - canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable); + + String[] destinations = StringUtils.split(System.getProperty("canal.destinations"), ","); + for (String destination : destinations) { + destination = destination.trim(); + CanalMQRunnable canalMQRunnable = new CanalMQRunnable(destination); + canalMQWorks.put(destination, canalMQRunnable); executorService.execute(canalMQRunnable); } @@ -95,7 +99,7 @@ public class CanalMQStarter { CanalInstance canalInstance = canalServer.getCanalInstances().get(destination); if (canalInstance != null) { stopDestination(destination); - CanalMQRunnable canalMQRunnable = new CanalMQRunnable(canalInstance); + CanalMQRunnable canalMQRunnable = new CanalMQRunnable(destination); canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable); executorService.execute(canalMQRunnable); logger.info("## Start the MQ work of destination:" + destination); @@ -111,15 +115,16 @@ public class CanalMQStarter { } } - private void worker(MQProperties.CanalDestination destination, AtomicBoolean destinationRunning) { + private void worker(String destination, AtomicBoolean destinationRunning) { while (!running || !destinationRunning.get()) ; - logger.info("## start the MQ producer: {}.", destination.getCanalDestination()); + logger.info("## start the MQ producer: {}.", destination); - final ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); + final ClientIdentity clientIdentity = new ClientIdentity(destination, (short) 1001, ""); while (running && destinationRunning.get()) { try { - if (!canalServer.getCanalInstances().containsKey(clientIdentity.getDestination())) { + CanalInstance canalInstance = canalServer.getCanalInstances().get(destination); + if (canalInstance == null) { try { Thread.sleep(3000); } catch (InterruptedException e) { @@ -127,8 +132,16 @@ public class CanalMQStarter { } continue; } + MQProperties.CanalDestination canalDestination = new MQProperties.CanalDestination(); + canalDestination.setCanalDestination(destination); + CanalMQConfig mqConfig = canalInstance.getMqConfig(); + canalDestination.setTopic(mqConfig.getTopic()); + canalDestination.setPartition(mqConfig.getPartition()); + canalDestination.setPartitionsNum(mqConfig.getPartitionsNum()); + canalDestination.setPartitionHash(mqConfig.getPartitionHashProperties()); + canalServer.subscribe(clientIdentity); - logger.info("## the MQ producer: {} is running now ......", destination.getCanalDestination()); + logger.info("## the MQ producer: {} is running now ......", destination); Long getTimeout = properties.getCanalGetTimeout(); int getBatchSize = properties.getCanalBatchSize(); @@ -145,7 +158,7 @@ public class CanalMQStarter { try { int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); if (batchId != -1 && size != 0) { - canalMQProducer.send(destination, message, new CanalMQProducer.Callback() { + canalMQProducer.send(canalDestination, message, new CanalMQProducer.Callback() { @Override public void commit() { @@ -177,23 +190,16 @@ public class CanalMQStarter { private class CanalMQRunnable implements Runnable { - private CanalInstance canalInstance; + private String destination; - public CanalMQRunnable(CanalInstance canalInstance){ - this.canalInstance = canalInstance; + CanalMQRunnable(String destination){ + this.destination = destination; } private AtomicBoolean running = new AtomicBoolean(true); @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); }