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

Merge branch 'master' into master

Browse Source
This commit is contained in:
rewerma
2019-07-14 19:24:42 +08:00
committed by GitHub
parent c070757da0 483044916d
commit 7cec0994e9
181 changed files with 23177 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
deployer/src/main/bin/startup.bat
View File

@@ -1,25 +1,25 @@
@echo off
@if not "%ECHO%" == "" echo %ECHO%
@if "%OS%" == "Windows_NT" setlocal
set ENV_PATH=.\
if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0%
set conf_dir=%ENV_PATH%\..\conf
set canal_conf=%conf_dir%\canal.properties
set logback_configurationFile=%conf_dir%\logback.xml
set CLASSPATH=%conf_dir%
set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH%
set JAVA_MEM_OPTS= -Xms128m -Xmx512m -XX:PermSize=128m
set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8
set JAVA_DEBUG_OPT= -server -Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=9099,server=y,suspend=n
set CANAL_OPTS= -DappName=otter-canal -Dlogback.configurationFile="%logback_configurationFile%" -Dcanal.conf="%canal_conf%"
set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %JAVA_DEBUG_OPT% %CANAL_OPTS%
set CMD_STR= java %JAVA_OPTS% -classpath "%CLASSPATH%" java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.deployer.CanalLauncher
echo start cmd : %CMD_STR%
@echo off
@if not "%ECHO%" == "" echo %ECHO%
@if "%OS%" == "Windows_NT" setlocal
set ENV_PATH=.\
if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0%
set conf_dir=%ENV_PATH%\..\conf
set canal_conf=%conf_dir%\canal.properties
set logback_configurationFile=%conf_dir%\logback.xml
set CLASSPATH=%conf_dir%
set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH%
set JAVA_MEM_OPTS= -Xms128m -Xmx512m -XX:PermSize=128m
set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8
set JAVA_DEBUG_OPT= -server -Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,address=9099,server=y,suspend=n
set CANAL_OPTS= -DappName=otter-canal -Dlogback.configurationFile="%logback_configurationFile%" -Dcanal.conf="%canal_conf%"
set JAVA_OPTS= %JAVA_MEM_OPTS% %JAVA_OPTS_EXT% %JAVA_DEBUG_OPT% %CANAL_OPTS%
set CMD_STR= java %JAVA_OPTS% -classpath "%CLASSPATH%" java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.deployer.CanalLauncher
echo start cmd : %CMD_STR%
java %JAVA_OPTS% -classpath "%CLASSPATH%" com.alibaba.otter.canal.deployer.CanalLauncher

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
View File

@@ -4,7 +4,7 @@ import java.text.MessageFormat;
/**
* 启动常用变量
*
*
* @author jianghang 2012-11-8 下午03:15:55
* @version 1.0.0
*/
@@ -16,6 +16,7 @@ public class CanalConstants {
public static final String CANAL_IP = ROOT + "." + "ip";
public static final String CANAL_PORT = ROOT + "." + "port";
public static final String CANAL_METRICS_PULL_PORT = ROOT + "." + "metrics.pull.port";
public static final String CANAL_ADMIN_JMX_PORT = ROOT + "." + "admin.jmx.port";
public static final String CANAL_ZKSERVERS = ROOT + "." + "zkServers";
public static final String CANAL_WITHOUT_NETTY = ROOT + "." + "withoutNetty";
@@ -48,6 +49,7 @@ public class CanalConstants {
public static final String CANAL_MQ_FLATMESSAGE = ROOT + "." + "mq.flatMessage";
public static final String CANAL_MQ_COMPRESSION_TYPE = ROOT + "." + "mq.compressionType";
public static final String CANAL_MQ_ACKS = ROOT + "." + "mq.acks";
public static final String CANAL_MQ_TRANSACTION = ROOT + "." + "mq.transaction";
public static final String CANAL_MQ_PRODUCERGROUP = ROOT + "." + "mq.producerGroup";
public static final String CANAL_ALIYUN_ACCESSKEY = ROOT + "." + "aliyun.accessKey";
public static final String CANAL_ALIYUN_SECRETKEY = ROOT + "." + "aliyun.secretKey";

8
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
View File

@@ -556,4 +556,12 @@ public class CanalController {
public void setCanalMQStarter(CanalMQStarter canalMQStarter) {
this.canalMQStarter = canalMQStarter;
}
public Map<InstanceMode, InstanceConfigMonitor> getInstanceConfigMonitors() {
return instanceConfigMonitors;
}
public Map<String, InstanceConfig> getInstanceConfigs() {
return instanceConfigs;
}
}

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

@@ -2,7 +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.CanalServerBean;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -19,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();
@@ -53,8 +56,8 @@ public class CanalLauncher {
}
}
final CanalStater canalStater = new CanalStater();
canalStater.start(properties);
final CanalStater canalStater = new CanalStater(properties);
canalStater.start();
if (remoteConfigLoader != null) {
remoteConfigLoader.startMonitor(new RemoteCanalConfigMonitor() {
@@ -63,8 +66,9 @@ public class CanalLauncher {
public void onChange(Properties properties) {
try {
// 远程配置canal.properties修改重新加载整个应用
canalStater.destroy();
canalStater.start(properties);
canalStater.stop();
canalStater.setProperties(properties);
canalStater.start();
} catch (Throwable throwable) {
logger.error(throwable.getMessage(), throwable);
}
@@ -72,8 +76,20 @@ public class CanalLauncher {
});
}
while (running) {
Thread.sleep(1000);
CanalServerAgent canalServerAgent = null;
String jmxPort = properties.getProperty(CanalConstants.CANAL_ADMIN_JMX_PORT);
if (StringUtils.isNotEmpty(jmxPort)) {
String ip = properties.getProperty(CanalConstants.CANAL_IP);
CanalServerMXBean canalServerMBean = new CanalServerBean(canalStater);
canalServerAgent = new CanalServerAgent(ip, Integer.parseInt(jmxPort), canalServerMBean);
Thread agentThread = new Thread(canalServerAgent::start);
agentThread.start();
}
runningLatch.await();
if (canalServerAgent != null) {
canalServerAgent.stop();
}
if (remoteConfigLoader != null) {

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

@@ -33,14 +33,35 @@ public class CanalStater {
private CanalMQProducer canalMQProducer = null;
private Thread shutdownThread = null;
private CanalMQStarter canalMQStarter = null;
private volatile Properties properties;
private volatile boolean running = false;
public CanalStater(Properties properties){
this.properties = properties;
}
public boolean isRunning() {
return running;
}
public Properties getProperties() {
return properties;
}
public void setProperties(Properties properties) {
this.properties = properties;
}
public CanalController getController() {
return controller;
}
/**
* 启动方法
*
* @param properties canal.properties 配置
* @throws Throwable
*/
synchronized void start(Properties properties) throws Throwable {
public synchronized void start() throws Throwable {
String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
if (serverMode.equalsIgnoreCase("kafka")) {
canalMQProducer = new CanalKafkaProducer();
@@ -97,7 +118,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 {
@@ -114,6 +135,8 @@ public class CanalStater {
canalMQStarter.start(mqProperties);
controller.setCanalMQStarter(canalMQStarter);
}
running = true;
}
/**
@@ -121,7 +144,7 @@ public class CanalStater {
*
* @throws Throwable
*/
synchronized void destroy() throws Throwable {
public synchronized void stop() throws Throwable {
if (controller != null) {
controller.stop();
controller = null;
@@ -135,6 +158,7 @@ public class CanalStater {
canalMQStarter = null;
canalMQProducer = null;
}
running = false;
}
/**

73
deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerAgent.java Normal file
View File

@@ -0,0 +1,73 @@
package com.alibaba.otter.canal.deployer.mbean;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.rmi.registry.LocateRegistry;
import javax.management.MBeanServer;
import javax.management.Notification;
import javax.management.NotificationListener;
import javax.management.ObjectName;
import javax.management.remote.JMXConnectorServer;
import javax.management.remote.JMXConnectorServerFactory;
import javax.management.remote.JMXServiceURL;
public class CanalServerAgent implements NotificationListener {
private static final Logger logger = LoggerFactory.getLogger(CanalServerBean.class);
private MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
private String ip = "0.0.0.0";
private int port;
private CanalServerMXBean canalServerMbean;
private volatile JMXConnectorServer cs;
public CanalServerAgent(String ip, int port, CanalServerMXBean canalServerMbean){
if (StringUtils.isNotEmpty(ip)) {
this.ip = ip;
}
this.port = port;
this.canalServerMbean = canalServerMbean;
}
@Override
public void handleNotification(Notification notification, Object handback) {
}
public synchronized void start() {
try {
if (cs == null) {
ObjectName name = new ObjectName("CanalServerAgent:type=CanalServerStatus");
mBeanServer.registerMBean(canalServerMbean, name);
LocateRegistry.createRegistry(port);
JMXServiceURL jmxServiceURL = new JMXServiceURL(
"service:jmx:rmi:///jndi/rmi://" + ip + ":" + port + "/jmxrmi");
cs = JMXConnectorServerFactory.newJMXConnectorServer(jmxServiceURL, null, mBeanServer);
cs.start();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
public synchronized void stop() {
if (cs != null) {
try {
cs.stop();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}

150
deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerBean.java Normal file
View File

@@ -0,0 +1,150 @@
package com.alibaba.otter.canal.deployer.mbean;
import com.alibaba.otter.canal.common.utils.FileUtils;
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 javax.management.NotificationBroadcasterSupport;
import java.util.Map;
public class CanalServerBean extends NotificationBroadcasterSupport implements CanalServerMXBean {
private static final Logger logger = LoggerFactory.getLogger(CanalServerBean.class);
private volatile int status;
private CanalStater canalStater;
public CanalServerBean(CanalStater canalStater){
this.canalStater = canalStater;
this.status = canalStater.isRunning() ? 1 : 0;
}
@Override
public int getStatus() {
return status;
}
@Override
public synchronized boolean start() {
try {
if (!canalStater.isRunning()) {
canalStater.start();
status = 1;
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean stop() {
try {
if (canalStater.isRunning()) {
canalStater.stop();
status = 0;
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized boolean restart() {
stop();
return start();
}
@Override
public synchronized boolean exit() {
stop();
CanalLauncher.runningLatch.countDown();
return true;
}
@Override
public synchronized boolean startInstance(String destination) {
try {
InstanceAction instanceAction = getInstanceAction(destination);
if (instanceAction != null) {
instanceAction.start(destination);
return true;
}
} 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);
return true;
}
} 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);
return true;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return false;
}
@Override
public synchronized String getRunningInstances() {
try {
Map<String, InstanceConfig> instanceConfigs = canalStater.getController().getInstanceConfigs();
if (instanceConfigs != null) {
return Joiner.on(",").join(instanceConfigs.keySet());
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return "";
}
@Override
public String canalLog() {
return FileUtils.readFileFromOffset("../logs/canal/canal.log", 100, "UTF-8");
}
@Override
public String instanceLog(String destination) {
return FileUtils.readFileFromOffset("../logs/" + destination + "/" + destination + ".log", 100, "UTF-8");
}
private InstanceAction getInstanceAction(String destination) {
Map<InstanceConfig.InstanceMode, InstanceConfigMonitor> monitors = canalStater.getController()
.getInstanceConfigMonitors();
SpringInstanceConfigMonitor monitor = (SpringInstanceConfigMonitor) monitors
.get(InstanceConfig.InstanceMode.SPRING);
Map<String, InstanceAction> instanceActions = monitor.getActions();
return instanceActions.get(destination);
}
}

26
deployer/src/main/java/com/alibaba/otter/canal/deployer/mbean/CanalServerMXBean.java Normal file
View File

@@ -0,0 +1,26 @@
package com.alibaba.otter.canal.deployer.mbean;
public interface CanalServerMXBean {
int getStatus();
boolean start();
boolean stop();
boolean restart();
boolean exit();
boolean startInstance(String destination);
boolean stopInstance(String destination);
boolean reloadInstance(String destination);
String getRunningInstances();
String canalLog();
String instanceLog(String destination);
}

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java
View File

@@ -50,6 +50,10 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
private volatile boolean isFirst = true;
public Map<String, InstanceAction> getActions() {
return actions;
}
public void start() {
super.start();
Assert.notNull(rootConf, "root conf dir is null!");

3
deployer/src/main/resources/canal.properties
View File

@@ -8,6 +8,7 @@ canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.admin.jmx.port = 11113
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
@@ -127,4 +128,4 @@ canal.mq.accessChannel = local
##################################################
canal.mq.kafka.kerberos.enable = false
canal.mq.kafka.kerberos.krb5FilePath = "../conf/kerberos/krb5.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"
canal.mq.kafka.kerberos.jaasFilePath = "../conf/kerberos/jaas.conf"