From 115990db100f4e2d469fe6bf169fa01cc015d5d3 Mon Sep 17 00:00:00 2001 From: zhangxin10 Date: Mon, 14 Dec 2015 11:03:52 +0800 Subject: [PATCH] =?UTF-8?q?1.=20=E7=A7=BB=E9=99=A4ZookeeperUtil=E7=B1=BB?= =?UTF-8?q?=E4=B8=AD=E7=9A=84=E5=BC=82=E5=B8=B8=EF=BC=8C=E5=9B=A0=E4=B8=BA?= =?UTF-8?q?=E5=9C=A8=E5=8D=8F=E8=B0=83=E7=BA=BF=E7=A8=8B=E4=B8=AD=E6=9C=89?= =?UTF-8?q?=E4=B8=AA=E5=A4=A7=E7=9A=84Try...catch=E6=9D=A5=E8=BF=9B?= =?UTF-8?q?=E8=A1=8C=E6=8E=A7=E5=88=B6=E9=87=8D=E5=81=9A=202.=20=E4=BF=AE?= =?UTF-8?q?=E5=A4=8DAlarmMessageDAO=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- skywalking-alarm/pom.xml | 85 ++++ .../alarm/AlarmMessageProcessThread.java | 8 +- .../skywalking/alarm/UserInfoCoordinator.java | 398 +++++++++--------- .../cloud/skywalking/alarm/conf/Config.java | 3 +- .../skywalking/alarm/dao/AlarmMessageDao.java | 4 +- .../skywalking/alarm/util/ProcessUtil.java | 2 +- .../cloud/skywalking/alarm/util/ZKUtil.java | 27 +- 7 files changed, 305 insertions(+), 222 deletions(-) diff --git a/skywalking-alarm/pom.xml b/skywalking-alarm/pom.xml index 549781f4e..1a0056a3f 100644 --- a/skywalking-alarm/pom.xml +++ b/skywalking-alarm/pom.xml @@ -75,6 +75,91 @@ + + org.apache.maven.plugins + maven-resources-plugin + 2.6 + + + copy-resources + install + + copy-resources + + + ${project.build.directory}/installer/config + + + src/main/resources + false + + + + + + copy-start-script + install + + copy-resources + + + ${project.build.directory}/installer/bin + + + bin + false + + + + + + + + maven-dependency-plugin + + + copy-dependencies + pre-integration-test + + copy-dependencies + + + + + ${project.build.directory}/installer/lib + false + true + + + + maven-jar-plugin + + + *.properties + *.xml + + sky-walking-server + ${project.build.directory}/installer/lib + + + + maven-antrun-plugin + + + clean + package + + run + + + + + + + + + + maven-compiler-plugin diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/AlarmMessageProcessThread.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/AlarmMessageProcessThread.java index 9c45e3d44..d1537dbb6 100644 --- a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/AlarmMessageProcessThread.java +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/AlarmMessageProcessThread.java @@ -56,6 +56,7 @@ public class AlarmMessageProcessThread extends Thread { //检查是否分配线程的状态(重新分配状态) if (status == ProcessThreadStatus.REDISTRIBUTING) { + logger.info("The current thread[{}] state will change for the idle state", currentThread().getName()); // 修改自身状态：(空闲状态) status = ProcessThreadStatus.FREE; ProcessUtil.changeProcessThreadStatus(threadId, ProcessThreadStatus.FREE); @@ -66,6 +67,7 @@ public class AlarmMessageProcessThread extends Thread { //检查分配线程的状态(分配完成状态) if (status == ProcessThreadStatus.REDISTRIBUTE_SUCCESS) { + logger.info("The current thread[{}] state will change for the busy", currentThread().getName()); // 获取待处理的用户 processUserIds = acquireProcessedUsers(); @@ -106,7 +108,7 @@ public class AlarmMessageProcessThread extends Thread { } } - private List acquireProcessedUsers() { + private List acquireProcessedUsers() throws Exception { String path = Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId; String value = ZKUtil.getPathData(path); ProcessThreadValue processThreadValue = new Gson().fromJson(value, ProcessThreadValue.class); @@ -134,7 +136,7 @@ public class AlarmMessageProcessThread extends Thread { private class CoordinatorStatusWatcher implements CuratorWatcher { @Override - public void process(WatchedEvent watchedEvent) { + public void process(WatchedEvent watchedEvent) throws Exception { if (watchedEvent.getType() == Watcher.Event.EventType.NodeDataChanged) { String value = ZKUtil.getPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId); ProcessThreadValue processThreadValue = new Gson().fromJson(value, ProcessThreadValue.class); @@ -144,7 +146,7 @@ public class AlarmMessageProcessThread extends Thread { try { ZKUtil.getPathDataWithWatch(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId, watcher); } catch (Exception e) { - e.printStackTrace(); + logger.error("Failed to Watcher path [{}]", e); } } } diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/UserInfoCoordinator.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/UserInfoCoordinator.java index f3858bbd2..ce8d715cf 100644 --- a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/UserInfoCoordinator.java +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/UserInfoCoordinator.java @@ -14,230 +14,240 @@ import org.apache.logging.log4j.Logger; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Set; +import java.util.*; import java.util.concurrent.TimeUnit; public class UserInfoCoordinator extends Thread { - private Logger logger = LogManager.getLogger(UserInfoCoordinator.class); + private Logger logger = LogManager.getLogger(UserInfoCoordinator.class); - private boolean redistributing; - private RegisterServerWatcher watcher = new RegisterServerWatcher(); - private InterProcessMutex lock = new InterProcessMutex( - ZKUtil.getZkClient(), Config.ZKPath.COORDINATOR_PATH); - private boolean isCoordinator = false; + private boolean redistributing; + private RegisterServerWatcher watcher = new RegisterServerWatcher(); + private InterProcessMutex lock = new InterProcessMutex( + ZKUtil.getZkClient(), Config.ZKPath.COORDINATOR_PATH); + private boolean isCoordinator = false; - public UserInfoCoordinator() { - } + public UserInfoCoordinator() { + super("UserInfoCoordinator"); + } - @Override - public void run() { - while (true) { - try { - if (!isCoordinator) { - while (!retryBecomeCoordinator()) { - try { - Thread.sleep(Config.Coordinator.RETRY_BECOME_COORDINATOR_WAIT_TIME); - } catch (Exception e) { - logger.error("Sleep Failed.", e); - } - } - - isCoordinator = true; - watcherRegisterServerPath(); - redistributing = true; - } + @Override + public void run() { + while (true) { + try { + if (!isCoordinator) { + logger.info("Begin to retry become a coordinator...."); + while (!retryBecomeCoordinator()) { + try { + Thread.sleep(Config.Coordinator.RETRY_BECOME_COORDINATOR_WAIT_TIME); + } catch (Exception e) { + logger.error("Sleep Failed.", e); + } + } + logger.info("Become a coordinator."); + isCoordinator = true; + watcherRegisterServerPath(); + redistributing = true; + } - // 检查是否有新服务注册或者在重分配过程做有新处理线程启动了 - if (!redistributing) { - try { - Thread.sleep(Config.Coordinator.CHECK_REDISTRIBUTE_INTERVAL); - } catch (InterruptedException e) { - logger.error("Sleep error", e); - } + // 检查是否有新服务注册或者在重分配过程做有新处理线程启动了 + if (!redistributing) { + try { + Thread.sleep(Config.Coordinator.CHECK_REDISTRIBUTE_INTERVAL); + } catch (InterruptedException e) { + logger.error("Sleep error", e); + } - continue; - } + continue; + } - redistributing = false; + redistributing = false; + // 获取当前所有的注册的处理线程 + List registeredThreads = acquireAllRegisteredThread(); + logger.info("Query a total of {} processing threads", registeredThreads.size()); + // 修改状态 (开始重新分配状态） + changeStatus(registeredThreads, + ProcessThreadStatus.REDISTRIBUTING); + // 检查所有的服务是否都处于空闲状态 + int retryTimes = 0; + while (!checkAllProcessStatus(registeredThreads, + ProcessThreadStatus.FREE)) { + try { + Thread.sleep(Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL); + retryTimes++; + } catch (InterruptedException e) { + logger.error("Sleep failed", e); + } - // 获取当前所有的注册的处理线程 - List registeredThreads = acquireAllRegisteredThread(); - // 修改状态 (开始重新分配状态） - changeStatus(registeredThreads, - ProcessThreadStatus.REDISTRIBUTING); - // 检查所有的服务是否都处于空闲状态 - int retryTimes = 0; - while (!checkAllProcessStatus(registeredThreads, - ProcessThreadStatus.FREE)) { - try { - Thread.sleep(Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL); - retryTimes++; - } catch (InterruptedException e) { - logger.error("Sleep failed", e); - } - - if(retryTimes > 1000){ - logger.warn("checking all processors are free, waiting {}ms", Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL * retryTimes); - retryTimes = 0; - } - } + if (retryTimes > 1000) { + logger.warn("checking all processors are free, waiting {}ms", Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL * retryTimes); + retryTimes = 0; + } + } - // 查询当前有多少用户 - List users = AlarmMessageDao.selectAllUserIds(); + // 查询当前有多少用户 + List users = AlarmMessageDao.selectAllUserIds(); + logger.info("Query a total of {} process user", users.size()); + // 将用户重新分配给服务 + List realRedistributeThread = allocationUser( + registeredThreads, users); + logger.info("Assign the user to be processed to {} processing threads", realRedistributeThread.size()); + // 修改状态(分配完成) + changeStatus(realRedistributeThread, + ProcessThreadStatus.REDISTRIBUTE_SUCCESS); + logger.info("Change the state of {} processing threads to be busy", realRedistributeThread.size()); + // 检查所有的服务是否都处于忙碌状态 + while (!checkAllProcessStatus(realRedistributeThread, + ProcessThreadStatus.BUSY)) { + try { + Thread.sleep(Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL); + } catch (InterruptedException e) { + logger.error("Sleep failed", e); + } - // 将用户重新分配给服务 - List realRedistributeThread = allocationUser( - registeredThreads, users); + if (retryTimes > 1000) { + logger.warn("checking all processors are busy, waiting {}ms", Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL * retryTimes); + retryTimes = 0; + } + } - // 修改状态(分配完成) - changeStatus(realRedistributeThread, - ProcessThreadStatus.REDISTRIBUTE_SUCCESS); + } catch (Exception e) { + logger.error("Failed to coordinate, retry. ", e); + releaseCoordinator(); + isCoordinator = false; + } + } + } - // 检查所有的服务是否都处于忙碌状态 - while (!checkAllProcessStatus(realRedistributeThread, - ProcessThreadStatus.BUSY)) { - try { - Thread.sleep(Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL); - } catch (InterruptedException e) { - logger.error("Sleep failed", e); - } - - if(retryTimes > 1000){ - logger.warn("checking all processors are busy, waiting {}ms", Config.Coordinator.CHECK_ALL_PROCESS_THREAD_INTERVAL * retryTimes); - retryTimes = 0; - } - } + private boolean retryBecomeCoordinator() { + try { + return lock.acquire( + Config.Coordinator.RETRY_GET_COORDINATOR_LOCK_INTERVAL, + TimeUnit.SECONDS); + } catch (Exception e) { + logger.error("Failed to acquire lock .", e); + return false; + } + } - } catch (Exception e) { - logger.error("Failed to coordinate, retry. ", e); - releaseCoordinator(); - isCoordinator = false; - } - } - } + private void releaseCoordinator() { + if (lock != null && lock.isAcquiredInThisProcess()) { + try { + lock.release(); + } catch (Exception e1) { + logger.error("Failed to release lock.", e1); + } + } + } - private boolean retryBecomeCoordinator() { - try { - return lock.acquire( - Config.Coordinator.RETRY_GET_COORDINATOR_LOCK_INTERVAL, - TimeUnit.SECONDS); - } catch (Exception e) { - logger.error("Failed to acquire lock .", e); - return false; - } - } + private List allocationUser(List registeredThreads, + List userIds) throws Exception { + List realRedistributeThread = new ArrayList(); + Set sortThreadIds = new HashSet(registeredThreads); + int step = (int) Math.ceil(userIds.size() * 1.0 / sortThreadIds.size()); + int start = 0; + int end = step; - private void releaseCoordinator() { - if (lock != null && lock.isAcquiredInThisProcess()) { - try { - lock.release(); - } catch (Exception e1) { - logger.error("Failed to release lock.", e1); - } - } - } + if (end > userIds.size()) { + end = userIds.size(); + } - private List allocationUser(List registeredThreads, - List userIds) { - List realRedistributeThread = new ArrayList(); - Set sortThreadIds = new HashSet(registeredThreads); - int step = (int) Math.ceil(userIds.size() * 1.0 / sortThreadIds.size()); - int start = 0; - int end = step; + for (String thread : sortThreadIds) { + if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" + + thread)) + continue; + String value = ZKUtil + .getPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/" + + thread); + ProcessThreadValue value1 = new Gson().fromJson(value, + ProcessThreadValue.class); + value1.setDealUserIds(userIds.subList(start, end)); + ZKUtil.setPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/" + + thread, new Gson().toJson(value1)); + // 实际重新分配的线程Id + realRedistributeThread.add(thread); - if (end > userIds.size()) { - end = userIds.size(); - } + start = end; + end += step; + if (start >= userIds.size()) { + break; + } + if (end > userIds.size()) { + end = userIds.size(); + } - for (String thread : sortThreadIds) { - if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" - + thread)) - continue; - String value = ZKUtil - .getPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/" - + thread); - ProcessThreadValue value1 = new Gson().fromJson(value, - ProcessThreadValue.class); - value1.setDealUserIds(userIds.subList(start, end)); - ZKUtil.setPathData(Config.ZKPath.REGISTER_SERVER_PATH + "/" - + thread, new Gson().toJson(value1)); - // 实际重新分配的线程Id - realRedistributeThread.add(thread); + } + return realRedistributeThread; + } - start = end; - end += step; - if (start >= userIds.size()) { - break; - } - if (end > userIds.size()) { - end = userIds.size(); - } + private List acquireAllRegisteredThread() throws Exception { + return ZKUtil.getChildren(Config.ZKPath.REGISTER_SERVER_PATH); + } - } - return realRedistributeThread; - } + private boolean checkAllProcessStatus(List registeredThreadIds, + ProcessThreadStatus status) throws Exception { + String registerPathPrefix = Config.ZKPath.REGISTER_SERVER_PATH + "/"; + for (String threadId : registeredThreadIds) { - private List acquireAllRegisteredThread() { - return ZKUtil.getChildren(Config.ZKPath.REGISTER_SERVER_PATH); - } + if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" + + threadId)) + continue; - private boolean checkAllProcessStatus(List registeredThreadIds, - ProcessThreadStatus status) { - String registerPathPrefix = Config.ZKPath.REGISTER_SERVER_PATH + "/"; - for (String threadId : registeredThreadIds) { + if (getProcessThreadStatus(registerPathPrefix, threadId) != status) { + return false; + } + } + return true; + } - if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" - + threadId)) - continue; + private ProcessThreadStatus getProcessThreadStatus( + String registerPathPrefix, String threadId) throws Exception { + if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId)) + return ProcessThreadStatus.FREE; + String value = ZKUtil.getPathData(registerPathPrefix + threadId); + if (value == null || value.length() == 0) + return ProcessThreadStatus.FREE; + ProcessThreadValue value1 = new Gson().fromJson(value, + ProcessThreadValue.class); + return ProcessThreadStatus.convert(value1.getStatus()); + } - if (getProcessThreadStatus(registerPathPrefix, threadId) != status) { - return false; - } - } - return true; - } + private void changeStatus(List registeredThreadIds, + ProcessThreadStatus status) throws Exception { + Iterator threadIterator = registeredThreadIds.iterator(); + String threadId; + while (threadIterator.hasNext()) { + threadId = threadIterator.next(); + if (!checkProcessThreadIsOnline(threadId)) { + threadIterator.remove(); + } + ProcessUtil.changeProcessThreadStatus(threadId, status); + } + } - private ProcessThreadStatus getProcessThreadStatus( - String registerPathPrefix, String threadId) { - if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId)) - return ProcessThreadStatus.FREE; - String value = ZKUtil.getPathData(registerPathPrefix + threadId); - if (value == null || value.length() == 0) - return ProcessThreadStatus.FREE; - ProcessThreadValue value1 = new Gson().fromJson(value, - ProcessThreadValue.class); - return ProcessThreadStatus.convert(value1.getStatus()); - } + private boolean checkProcessThreadIsOnline(String threadId) { + return ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId); + } - private void changeStatus(List registeredThreadIds, - ProcessThreadStatus status) { - for (String threadId : registeredThreadIds) { - ProcessUtil.changeProcessThreadStatus(threadId, status); - } - } + public class RegisterServerWatcher implements CuratorWatcher { - public class RegisterServerWatcher implements CuratorWatcher { + @Override + public void process(WatchedEvent watchedEvent) { + if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) { + logger.info("The number of process thread has changed. The alarm service will reallocate processing data."); + redistributing = true; + } - @Override - public void process(WatchedEvent watchedEvent) { - if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) { - redistributing = true; - } + watcherRegisterServerPath(); + } + } - watcherRegisterServerPath(); - } - } - - private void watcherRegisterServerPath() { - try { - ZKUtil.getChildrenWithWatcher(Config.ZKPath.REGISTER_SERVER_PATH, - watcher); - } catch (Exception e) { - logger.error("Failed to set watcher for get children", e); - } - } + private void watcherRegisterServerPath() { + try { + ZKUtil.getChildrenWithWatcher(Config.ZKPath.REGISTER_SERVER_PATH, + watcher); + } catch (Exception e) { + logger.error("Failed to set watcher for get children", e); + } + } } diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/conf/Config.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/conf/Config.java index 68ae891a2..a466c4f0b 100644 --- a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/conf/Config.java +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/conf/Config.java @@ -11,7 +11,8 @@ public class Config { } public static class ProcessThread { - public static long THREAD_WAIT_INTERVAL = 60 * 1000L; + // public static long THREAD_WAIT_INTERVAL = 60 * 1000L; + public static long THREAD_WAIT_INTERVAL = 5 * 1000L; } public static class ZKPath { diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/dao/AlarmMessageDao.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/dao/AlarmMessageDao.java index 4fafab1a9..41af94f87 100644 --- a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/dao/AlarmMessageDao.java +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/dao/AlarmMessageDao.java @@ -86,8 +86,8 @@ public class AlarmMessageDao { continue; } else { tmpAlarmRule = new AlarmRule(rs.getString("uid"), rs.getString("rule_id")); - globalRules.setConfigArgs(rs.getString("config_args")); - globalRules.setTodoType(rs.getString("todo_type")); + tmpAlarmRule.setConfigArgs(rs.getString("config_args")); + tmpAlarmRule.setTodoType(rs.getString("todo_type")); // 自定义规则的Application tmpApplication = new ApplicationInfo(); tmpApplication.setAppId(rs.getString("app_id")); diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ProcessUtil.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ProcessUtil.java index df55e24e4..9d102db5b 100644 --- a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ProcessUtil.java +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ProcessUtil.java @@ -7,7 +7,7 @@ import com.google.gson.Gson; public class ProcessUtil { - public static void changeProcessThreadStatus(String threadId, ProcessThreadStatus status) { + public static void changeProcessThreadStatus(String threadId, ProcessThreadStatus status) throws Exception { String path = Config.ZKPath.REGISTER_SERVER_PATH + "/" + threadId; String value = ZKUtil.getPathData(path); ProcessThreadValue newValue = new Gson().fromJson(value, ProcessThreadValue.class); diff --git a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ZKUtil.java b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ZKUtil.java index 24f4a491a..bddbd7b81 100644 --- a/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ZKUtil.java +++ b/skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ZKUtil.java @@ -11,7 +11,6 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.zookeeper.CreateMode; -import java.util.ArrayList; import java.util.List; public class ZKUtil { @@ -42,34 +41,20 @@ public class ZKUtil { } - public static String getPathData(String path) { - try { - return new String(client.getData().forPath(path)); - } catch (Exception e) { - logger.error("Failed to get the value of path[{}]", path, e); - } - return ""; + public static String getPathData(String path) throws Exception { + return new String(client.getData().forPath(path)); } public static String getPathDataWithWatch(String path, CuratorWatcher watcher) throws Exception { return new String(client.getData().usingWatcher(watcher).forPath(path)); } - public static void setPathData(String path, String value) { - try { - client.setData().forPath(path, value.getBytes()); - } catch (Exception e) { - logger.error("Failed to set date of path[{{}]", path, e); - } + public static void setPathData(String path, String value) throws Exception { + client.setData().forPath(path, value.getBytes()); } - public static List getChildren(String registerServerPath) { - try { - return client.getChildren().forPath(registerServerPath); - } catch (Exception e) { - logger.error("Failed to get child nodes of path[{{}]", registerServerPath, e); - } - return new ArrayList(); + public static List getChildren(String registerServerPath) throws Exception { + return client.getChildren().forPath(registerServerPath); } public static List getChildrenWithWatcher(String registerServerPath, CuratorWatcher watcher) throws Exception {