1. CanalController stop 需要同时将 embededCanalServer.stop (#4477)

2. ServerRunningMonitor 线程池未正常回收,线程池管理与 start/stop保持一致
This commit is contained in:
华仔
2022-11-16 09:57:59 +08:00
committed by GitHub
parent 10ff148aba
commit fa0c08c170
2 changed files with 12 additions and 2 deletions

View File

@@ -38,7 +38,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
private ServerRunningData serverData; private ServerRunningData serverData;
// 当前实际运行的节点状态信息 // 当前实际运行的节点状态信息
private volatile ServerRunningData activeData; private volatile ServerRunningData activeData;
private ScheduledExecutorService delayExector = Executors.newScheduledThreadPool(1); private ScheduledExecutorService delayExecutor;
private int delayTime = 5; private int delayTime = 5;
private ServerRunningListener listener; private ServerRunningListener listener;
@@ -73,7 +73,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
initRunning(); initRunning();
} else { } else {
// 否则就是等待delayTime,避免因网络瞬端或者zk异常,导致出现频繁的切换操作 // 否则就是等待delayTime,避免因网络瞬端或者zk异常,导致出现频繁的切换操作
delayExector.schedule(() -> initRunning(), delayTime, TimeUnit.SECONDS); delayExecutor.schedule(() -> initRunning(), delayTime, TimeUnit.SECONDS);
} }
} }
@@ -90,6 +90,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
try { try {
processStart(); processStart();
if (zkClient != null) { if (zkClient != null) {
delayExecutor = Executors.newScheduledThreadPool(1);
// 如果需要尽可能释放instance资源,不需要监听running节点,不然即使stop了这台机器,另一台机器立马会start // 如果需要尽可能释放instance资源,不需要监听running节点,不然即使stop了这台机器,另一台机器立马会start
String path = ZookeeperPathUtils.getDestinationServerRunning(destination); String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
zkClient.subscribeDataChanges(path, dataListener); zkClient.subscribeDataChanges(path, dataListener);
@@ -122,6 +123,10 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
if (zkClient != null) { if (zkClient != null) {
String path = ZookeeperPathUtils.getDestinationServerRunning(destination); String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
zkClient.unsubscribeDataChanges(path, dataListener); zkClient.unsubscribeDataChanges(path, dataListener);
if (delayExecutor != null) {
delayExecutor.shutdown();
delayExecutor = null;
}
releaseRunning(); // 尝试一下release releaseRunning(); // 尝试一下release
} else { } else {

View File

@@ -558,6 +558,11 @@ public class CanalController {
} }
ZkClientx.clearClients(); ZkClientx.clearClients();
// 需要释放 CanalServerWithEmbedded 否则主线程退出后,进程无法自动完整退出...
if (embededCanalServer != null) {
embededCanalServer.stop();
}
} }
private void initCid(String path) { private void initCid(String path) {