From 5741f5a347d49a492b258b45c14b5ac70063c549 Mon Sep 17 00:00:00 2001 From: rewerma Date: Fri, 12 Jul 2019 00:42:18 +0800 Subject: [PATCH] modify --- .../otter/canal/deployer/CanalController.java | 8 ++ .../otter/canal/deployer/CanalStater.java | 4 + .../canal/deployer/mbean/CanalServerBean.java | 73 ++++++++++++++++++- .../deployer/mbean/CanalServerMXBean.java | 8 ++ .../monitor/SpringInstanceConfigMonitor.java | 4 + 5 files changed, 93 insertions(+), 4 deletions(-) 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 c0692227..6671287d 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 @@ -556,4 +556,12 @@ public class CanalController { public void setCanalMQStarter(CanalMQStarter canalMQStarter) { this.canalMQStarter = canalMQStarter; } + + public Map getInstanceConfigMonitors() { + return instanceConfigMonitors; + } + + public Map getInstanceConfigs() { + return instanceConfigs; + } } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java index aa2a9d3c..bf231e72 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java @@ -51,6 +51,10 @@ public class CanalStater { this.properties = properties; } + public CanalController getController() { + return controller; + } + /** * 启动方法 * diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java index d6fcdbde..c08a547b 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java @@ -2,9 +2,16 @@ package com.alibaba.otter.canal.deployer.mbean; import com.alibaba.otter.canal.deployer.CanalLauncher; import com.alibaba.otter.canal.deployer.CanalStater; +import com.alibaba.otter.canal.deployer.InstanceConfig; +import com.alibaba.otter.canal.deployer.monitor.InstanceAction; +import com.alibaba.otter.canal.deployer.monitor.InstanceConfigMonitor; +import com.alibaba.otter.canal.deployer.monitor.SpringInstanceConfigMonitor; +import com.google.common.base.Joiner; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Map; + public class CanalServerBean implements CanalServerMXBean { private static final Logger logger = LoggerFactory.getLogger(CanalServerBean.class); @@ -23,10 +30,6 @@ public class CanalServerBean implements CanalServerMXBean { return status; } - public void setStatus(int status) { - this.status = status; - } - @Override public synchronized boolean start() { try { @@ -68,4 +71,66 @@ public class CanalServerBean implements CanalServerMXBean { return true; } + @Override + public synchronized boolean startInstance(String destination) { + try { + InstanceAction instanceAction = getInstanceAction(destination); + if (instanceAction != null) { + instanceAction.start(destination); + + } + } catch (Throwable e) { + logger.error(e.getMessage(), e); + } + return false; + } + + @Override + public synchronized boolean stopInstance(String destination) { + try { + InstanceAction instanceAction = getInstanceAction(destination); + if (instanceAction != null) { + instanceAction.stop(destination); + } + } catch (Throwable e) { + logger.error(e.getMessage(), e); + } + return false; + } + + @Override + public synchronized boolean reloadInstance(String destination) { + try { + InstanceAction instanceAction = getInstanceAction(destination); + if (instanceAction != null) { + instanceAction.reload(destination); + } + } catch (Throwable e) { + logger.error(e.getMessage(), e); + } + return false; + } + + @Override + public synchronized String getRunningInstances() { + try { + Map instanceConfigs = canalStater.getController().getInstanceConfigs(); + if (instanceConfigs != null) { + return Joiner.on(",").join(instanceConfigs.keySet()); + } + } catch (Throwable e) { + logger.error(e.getMessage(), e); + } + return ""; + } + + private InstanceAction getInstanceAction(String destination) { + Map monitors = canalStater.getController() + .getInstanceConfigMonitors(); + SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors + .get(InstanceConfig.InstanceMode.SPRING); + Map instanceActions = monitor.getActions(); + return instanceActions.get(destination); + } + } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerMXBean.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerMXBean.java index 77a7e16c..0fad0a98 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerMXBean.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerMXBean.java @@ -11,4 +11,12 @@ public interface CanalServerMXBean { boolean restart(); boolean exit(); + + boolean startInstance(String destination); + + boolean stopInstance(String destination); + + boolean reloadInstance(String destination); + + String getRunningInstances(); } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java index 3483324d..b8cef713 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java @@ -50,6 +50,10 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen private volatile boolean isFirst = true; + public Map getActions() { + return actions; + } + public void start() { super.start(); Assert.notNull(rootConf, "root conf dir is null!");