chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

modify

Browse Source
This commit is contained in:
rewerma
2019-07-11 23:29:49 +08:00
parent 23209f24ed
commit 64824bc5aa
3 changed files with 11 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
View File

@@ -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();

2
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java
View File

@@ -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 {

6
deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServer.java → deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java
View File

@@ -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;
}