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

canal-admin完善 (#2121)

Browse Source 
* admin端口调整完成

* server端适配调整

* 调整部分功能

* 单机/集群admin调试完成

* 重新发布前端页面
This commit is contained in:
rewerma
2019-08-30 18:09:22 +08:00
committed by agapple
parent e8dba41216
commit 89055f7405
60 changed files with 1882 additions and 382 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -154,86 +154,87 @@ public class CanalController {
final ServerRunningData serverData = new ServerRunningData(cid, registerIp + ":" + port);
ServerRunningMonitors.setServerData(serverData);
ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
ServerRunningMonitors
.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
public ServerRunningMonitor apply(final String destination) {
ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
runningMonitor.setDestination(destination);
runningMonitor.setListener(new ServerRunningListener() {
public ServerRunningMonitor apply(final String destination) {
ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
runningMonitor.setDestination(destination);
runningMonitor.setListener(new ServerRunningListener() {
public void processActiveEnter() {
try {
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
embededCanalServer.start(destination);
if (canalMQStarter != null) {
canalMQStarter.startDestination(destination);
public void processActiveEnter() {
try {
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
embededCanalServer.start(destination);
if (canalMQStarter != null) {
canalMQStarter.startDestination(destination);
}
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
}
public void processActiveExit() {
try {
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
if (canalMQStarter != null) {
canalMQStarter.stopDestination(destination);
public void processActiveExit() {
try {
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
if (canalMQStarter != null) {
canalMQStarter.stopDestination(destination);
}
embededCanalServer.stop(destination);
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
embededCanalServer.stop(destination);
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
}
public void processStart() {
try {
if (zkclientx != null) {
final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
registerIp + ":" + port);
initCid(path);
zkclientx.subscribeStateChanges(new IZkStateListener() {
public void processStart() {
try {
if (zkclientx != null) {
final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
registerIp + ":" + port);
initCid(path);
zkclientx.subscribeStateChanges(new IZkStateListener() {
public void handleStateChanged(KeeperState state) throws Exception {
public void handleStateChanged(KeeperState state) throws Exception {
}
}
public void handleNewSession() throws Exception {
initCid(path);
}
public void handleNewSession() throws Exception {
initCid(path);
}
@Override
public void handleSessionEstablishmentError(Throwable error) throws Exception {
logger.error("failed to connect to zookeeper", error);
}
});
@Override
public void handleSessionEstablishmentError(Throwable error) throws Exception {
logger.error("failed to connect to zookeeper", error);
}
});
}
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
}
public void processStop() {
try {
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
if (zkclientx != null) {
final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
registerIp + ":" + port);
releaseCid(path);
public void processStop() {
try {
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
if (zkclientx != null) {
final String path = ZookeeperPathUtils.getDestinationClusterNode(destination,
registerIp + ":" + port);
releaseCid(path);
}
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
}
});
if (zkclientx != null) {
runningMonitor.setZkClient(zkclientx);
});
if (zkclientx != null) {
runningMonitor.setZkClient(zkclientx);
}
// 触发创建一下cid节点
runningMonitor.init();
return runningMonitor;
}
// 触发创建一下cid节点
runningMonitor.init();
return runningMonitor;
}
}));
}));
// 初始化monitor机制
autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
@@ -285,7 +286,8 @@ public class CanalController {
instanceConfigMonitors = MigrateMap.makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>() {
public InstanceConfigMonitor apply(InstanceMode mode) {
int scanInterval = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL));
int scanInterval = Integer
.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL));
if (mode.isSpring()) {
SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor();
@@ -381,7 +383,7 @@ public class CanalController {
}
private PlainCanalConfigClient getManagerClient(String managerAddress) {
return new PlainCanalConfigClient(managerAddress, this.adminUser, this.adminPasswd);
return new PlainCanalConfigClient(managerAddress, this.adminUser, this.adminPasswd, null, adminPort);
}
private void initInstanceConfig(Properties properties) {
@@ -393,7 +395,8 @@ public class CanalController {
InstanceConfig oldConfig = instanceConfigs.put(destination, config);
if (oldConfig != null) {
logger.warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config);
logger
.warn("destination:{} old config:{} has replace by new config:{}", destination, oldConfig, config);
}
}
}

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

@@ -27,7 +27,7 @@ public class CanalLauncher {
private static final Logger logger = LoggerFactory.getLogger(CanalLauncher.class);
public static final CountDownLatch runningLatch = new CountDownLatch(1);
private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
new NamedThreadFactory("canal-server-scan"));
new NamedThreadFactory("canal-server-scan"));
public static void main(String[] args) {
try {
@@ -49,11 +49,19 @@ public class CanalLauncher {
if (StringUtils.isNotEmpty(managerAddress)) {
String user = properties.getProperty(CanalConstants.CANAL_ADMIN_USER);
String passwd = properties.getProperty(CanalConstants.CANAL_ADMIN_PASSWD);
final PlainCanalConfigClient configClient = new PlainCanalConfigClient(managerAddress, user, passwd);
String adminPort = properties.getProperty(CanalConstants.CANAL_ADMIN_PORT);
if (StringUtils.isEmpty(adminPort)) {
adminPort = "11110";
}
final PlainCanalConfigClient configClient = new PlainCanalConfigClient(managerAddress,
user,
passwd,
"",
Integer.parseInt(adminPort));
PlainCanal canalConfig = configClient.findServer(null);
properties = canalConfig.getProperties();
int scanIntervalInSecond = Integer.valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL,
"5"));
int scanIntervalInSecond = Integer
.valueOf(properties.getProperty(CanalConstants.CANAL_AUTO_SCAN_INTERVAL, "5"));
executor.scheduleWithFixedDelay(new Runnable() {
private PlainCanal lastCanalConfig;

20
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java
View File

@@ -22,24 +22,26 @@ import com.google.common.collect.MigrateMap;
/**
* 基于manager配置的实现
*
*
* @author agapple 2019年8月26日 下午10:00:20
* @since 1.1.4
*/
public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, CanalLifeCycle {
private static final Logger logger = LoggerFactory.getLogger(ManagerInstanceConfigMonitor.class);
private static final Logger logger = LoggerFactory
.getLogger(ManagerInstanceConfigMonitor.class);
private long scanIntervalInSecond = 5;
private InstanceAction defaultAction = null;
private Map<String, InstanceAction> actions = new MapMaker().makeMap();
private Map<String, PlainCanal> configs = MigrateMap.makeComputingMap(new Function<String, PlainCanal>() {
private Map<String, PlainCanal> configs = MigrateMap
.makeComputingMap(new Function<String, PlainCanal>() {
public PlainCanal apply(String destination) {
return new PlainCanal();
}
});
public PlainCanal apply(String destination) {
return new PlainCanal();
}
});
private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
new NamedThreadFactory("canal-instance-scan"));
new NamedThreadFactory("canal-instance-scan"));
private volatile boolean isFirst = true;
private PlainCanalConfigClient configClient;
@@ -84,7 +86,7 @@ public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle impleme
}
private void scan() {
String instances = configClient.findInstances(ip, String.valueOf(port), lastInstanceMD5);
String instances = configClient.findInstances(lastInstanceMD5);
final List<String> is = Lists.newArrayList(StringUtils.split(instances, ','));
List<String> start = Lists.newArrayList();
List<String> stop = Lists.newArrayList();