修改mq配置针对destination级别重启

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java

@@ -226,7 +226,7 @@ public class CanalController {
                         if (!config.getLazy() && !runningMonitor.isStart()) {
                             runningMonitor.start();
                             if (canalMQStarter != null) {
-                                canalMQStarter.start(getMQProperties(properties));
+                                canalMQStarter.startDestination(destination);
                             }
                         }
                     }
@@ -237,7 +237,7 @@ public class CanalController {
                     InstanceConfig config = instanceConfigs.remove(destination);
                     if (config != null) {
                         if (canalMQStarter != null) {
-                            canalMQStarter.stop();
+                            canalMQStarter.stopDestination(destination);
                         }
                         embededCanalServer.stop(destination);
                         ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
@@ -533,24 +533,4 @@ public class CanalController {
     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;
-    }
 }

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java

@@ -3,6 +3,7 @@ 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;
@@ -72,7 +73,23 @@ public class CanalLauncher {
 
             if (canalMQProducer != null) {
                 CanalMQStarter canalMQStarter = new CanalMQStarter(canalMQProducer);
-                canalMQStarter.start(CanalController.getMQProperties(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)));
+                canalMQStarter.start(mqProperties);
                 controller.setCanalMQStarter(canalMQStarter);
             }
         } catch (Throwable e) {

server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java

@@ -1,8 +1,11 @@
 package com.alibaba.otter.canal.server;
 
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,6 +33,8 @@ public class CanalMQStarter {
 
     private CanalServerWithEmbedded      canalServer;
 
+    private Map<String, CanalMQRunnable> canalMQWorks = new ConcurrentHashMap<>();
+
     public CanalMQStarter(CanalMQProducer canalMQProducer){
         this.canalMQProducer = canalMQProducer;
     }
@@ -57,20 +62,9 @@ public class CanalMQStarter {
             executorService = Executors.newFixedThreadPool(canalServer.getCanalInstances().size());
             logger.info("## start the MQ workers.");
             for (final CanalInstance canalInstance : canalServer.getCanalInstances().values()) {
-                executorService.execute(new Runnable() {
+                CanalMQRunnable canalMQRunnable = new CanalMQRunnable(canalInstance);
-
+                canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable);
-                    @Override
+                executorService.execute(canalMQRunnable);
-                    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;
@@ -78,21 +72,6 @@ public class CanalMQStarter {
             Runtime.getRuntime().addShutdownHook(new Thread() {
 
                 public void run() {
-                    stop();
-                }
-
-            });
-
-        } catch (Throwable e) {
-            logger.error("## Something goes wrong when starting up the canal MQ workers:", e);
-            System.exit(0);
-        }
-    }
-
-    public synchronized void stop() {
-        if (!running) {
-            return;
-        }
                     try {
                         logger.info("## stop the MQ workers");
                         running = false;
@@ -105,13 +84,41 @@ public class CanalMQStarter {
                     }
                 }
 
-    private void worker(MQProperties.CanalDestination destination) {
+            });
-        while (!running)
+
+        } catch (Throwable e) {
+            logger.error("## Something goes wrong when starting up the canal MQ workers:", e);
+            System.exit(0);
+        }
+    }
+
+    public synchronized void startDestination(String destination) {
+        CanalInstance canalInstance = canalServer.getCanalInstances().get(destination);
+        if (canalInstance != null) {
+            stopDestination(destination);
+            CanalMQRunnable canalMQRunnable = new CanalMQRunnable(canalInstance);
+            canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable);
+            executorService.execute(canalMQRunnable);
+            logger.info("## Start the MQ work of destination:" + destination);
+        }
+    }
+
+    public synchronized void stopDestination(String destination) {
+        CanalMQRunnable canalMQRunable = canalMQWorks.get(destination);
+        if (canalMQRunable != null) {
+            canalMQRunable.stop();
+            canalMQWorks.remove(destination);
+            logger.info("## Stop the MQ work of destination:" + destination);
+        }
+    }
+
+    private void worker(MQProperties.CanalDestination destination, AtomicBoolean destinationRunning) {
+        while (!running || !destinationRunning.get())
             ;
         logger.info("## start the MQ producer: {}.", destination.getCanalDestination());
 
         final ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, "");
-        while (running) {
+        while (running && destinationRunning.get()) {
             try {
                 if (!canalServer.getCanalInstances().containsKey(clientIdentity.getDestination())) {
                     try {
@@ -126,7 +133,7 @@ public class CanalMQStarter {
 
                 Long getTimeout = properties.getCanalGetTimeout();
                 int getBatchSize = properties.getCanalBatchSize();
-                while (running) {
+                while (running && destinationRunning.get()) {
                     Message message;
                     if (getTimeout != null && getTimeout > 0) {
                         message = canalServer
@@ -168,4 +175,31 @@ public class CanalMQStarter {
             }
         }
     }
+
+    private class CanalMQRunnable implements Runnable {
+
+        private CanalInstance canalInstance;
+
+        public CanalMQRunnable(CanalInstance canalInstance){
+            this.canalInstance = canalInstance;
+        }
+
+        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);
+        }
+
+        public void stop() {
+            running.set(false);
+        }
+    }
 }