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 ff5b121d..5da5d28b 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 @@ -2,10 +2,11 @@ package com.alibaba.otter.canal.deployer; import java.io.FileInputStream; import java.util.Properties; +import java.util.concurrent.CountDownLatch; import com.alibaba.otter.canal.deployer.mbean.CanalServerAgent; import com.alibaba.otter.canal.deployer.mbean.CanalServerMXBean; -import com.alibaba.otter.canal.deployer.mbean.CanalServer; +import com.alibaba.otter.canal.deployer.mbean.CanalServerBean; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -22,13 +23,12 @@ import com.alibaba.otter.canal.deployer.monitor.remote.RemoteCanalConfigMonitor; */ public class CanalLauncher { - private static final String CLASSPATH_URL_PREFIX = "classpath:"; - private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); - public static volatile boolean running = false; + private static final String CLASSPATH_URL_PREFIX = "classpath:"; + private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class); + public static final CountDownLatch runningLatch = new CountDownLatch(1); public static void main(String[] args) { try { - running = true; logger.info("## set default uncaught exception handler"); setGlobalUncaughtExceptionHandler(); @@ -80,15 +80,13 @@ public class CanalLauncher { String jmxPort = properties.getProperty(CanalConstants.CANAL_ADMIN_JMX_PORT); if (StringUtils.isNotEmpty(jmxPort)) { String ip = properties.getProperty(CanalConstants.CANAL_IP); - CanalServerMXBean canalServerMBean = new CanalServer(canalStater); + CanalServerMXBean canalServerMBean = new CanalServerBean(canalStater); canalServerAgent = new CanalServerAgent(ip, Integer.parseInt(jmxPort), canalServerMBean); Thread agentThread = new Thread(canalServerAgent::start); agentThread.start(); } - while (running) { - Thread.sleep(1000); - } + runningLatch.await(); if (canalServerAgent != null) { canalServerAgent.stop(); 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 c4ba632c..aa2a9d3c 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 @@ -113,7 +113,7 @@ public class CanalStater { try { logger.info("## stop the canal server"); controller.stop(); - CanalLauncher.running = false; + CanalLauncher.runningLatch.countDown(); } catch (Throwable e) { logger.warn("##something goes wrong when stopping canal Server:", e); } finally { diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServer.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java similarity index 89% rename from deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServer.java rename to deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java index 88c94e23..824c883c 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServer.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java @@ -3,13 +3,13 @@ package com.alibaba.otter.canal.deployer.mbean; import com.alibaba.otter.canal.deployer.CanalLauncher; import com.alibaba.otter.canal.deployer.CanalStater; -public class CanalServer implements CanalServerMXBean { +public class CanalServerBean implements CanalServerMXBean { private volatile int status; private CanalStater canalStater; - public CanalServer(CanalStater canalStater){ + public CanalServerBean(CanalStater canalStater){ this.canalStater = canalStater; this.status = canalStater.isRunning() ? 1 : 0; } @@ -60,7 +60,7 @@ public class CanalServer implements CanalServerMXBean { @Override public synchronized boolean exit() { stop(); - CanalLauncher.running = false; + CanalLauncher.runningLatch.countDown(); return true; }