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

增加注释

Browse Source
This commit is contained in:
machey
2018-12-30 17:08:11 +08:00
parent 661752096c
commit f9e19d46c4
1 changed files with 25 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,6 +12,12 @@ import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer;
import com.alibaba.otter.canal.server.CanalMQStarter; import com.alibaba.otter.canal.server.CanalMQStarter;
import com.alibaba.otter.canal.spi.CanalMQProducer; import com.alibaba.otter.canal.spi.CanalMQProducer;
/**
* Canal server 启动类
*
* @author rewerma 2018-12-30 下午05:12:16
* @version 1.0.1
*/
public class CanalStater { public class CanalStater {
private static final Logger logger = LoggerFactory.getLogger(CanalStater.class); private static final Logger logger = LoggerFactory.getLogger(CanalStater.class);
@@ -21,6 +27,12 @@ public class CanalStater {
private Thread shutdownThread = null; private Thread shutdownThread = null;
private CanalMQStarter canalMQStarter = null; private CanalMQStarter canalMQStarter = null;
/**
* 启动方法
*
* @param properties canal.properties 配置
* @throws Throwable
*/
synchronized void start(Properties properties) throws Throwable { synchronized void start(Properties properties) throws Throwable {
String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE); String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
if (serverMode.equalsIgnoreCase("kafka")) { if (serverMode.equalsIgnoreCase("kafka")) {
@@ -59,12 +71,17 @@ public class CanalStater {
if (canalMQProducer != null) { if (canalMQProducer != null) {
canalMQStarter = new CanalMQStarter(canalMQProducer); canalMQStarter = new CanalMQStarter(canalMQProducer);
MQProperties mqProperties = buildMQPosition(properties); MQProperties mqProperties = buildMQProperties(properties);
canalMQStarter.start(mqProperties); canalMQStarter.start(mqProperties);
controller.setCanalMQStarter(canalMQStarter); controller.setCanalMQStarter(canalMQStarter);
} }
} }
/**
* 销毁方法，远程配置变更时调用
*
* @throws Throwable
*/
synchronized void destroy() throws Throwable { synchronized void destroy() throws Throwable {
if (controller != null) { if (controller != null) {
controller.stop(); controller.stop();
@@ -81,7 +98,13 @@ public class CanalStater {
} }
} }
private static MQProperties buildMQPosition(Properties properties) { /**
* 构造MQ对应的配置
*
* @param properties canal.properties 配置
* @return
*/
private static MQProperties buildMQProperties(Properties properties) {
MQProperties mqProperties = new MQProperties(); MQProperties mqProperties = new MQProperties();
String servers = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS); String servers = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_SERVERS);
if (!StringUtils.isEmpty(servers)) { if (!StringUtils.isEmpty(servers)) {