kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. 移除ZookeeperUtil类中的异常，因为在协调线程中有个大的Try...catch来进行控制重做 

2. 修复AlarmMessageDAO的问题
This commit is contained in:
zhangxin10 2015-12-14 11:03:52 +08:00
parent 2c1fb7913a
commit 115990db10
7 changed files with 305 additions and 222 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
skywalking-alarm/pom.xml
View File

@ -75,6 +75,91 @@
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<version>2.6</version>
<executions>
<execution>
<id>copy-resources</id>
<phase>install</phase>
<goals>
<goal>copy-resources</goal>
</goals>
<configuration>
<outputDirectory>${project.build.directory}/installer/config</outputDirectory>
<resources>
<resource>
<directory>src/main/resources</directory>
<filtering>false</filtering>
</resource>
</resources>
</configuration>
</execution>
<execution>
<id>copy-start-script</id>
<phase>install</phase>
<goals>
<goal>copy-resources</goal>
</goals>
<configuration>
<outputDirectory>${project.build.directory}/installer/bin</outputDirectory>
<resources>
<resource>
<directory>bin</directory>
<filtering>false</filtering>
</resource>
</resources>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<id>copy-dependencies</id>
<phase>pre-integration-test</phase>
<goals>
<goal>copy-dependencies</goal>
</goals>
</execution>
</executions>
<configuration>
<outputDirectory>${project.build.directory}/installer/lib</outputDirectory>
<excludeTransitive>false</excludeTransitive>
<stripVersion>true</stripVersion>
</configuration>
</plugin>
<plugin>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<excludes>
<exclude>*.properties</exclude>
<exclude>*.xml</exclude>
</excludes>
<finalName>sky-walking-server</finalName>
<outputDirectory>${project.build.directory}/installer/lib</outputDirectory>
</configuration>
</plugin>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<id>clean</id>
<phase>package</phase>
<goals>
<goal>run</goal>
</goals>
<configuration>
<tasks>
<mkdir dir="${project.build.directory}/installer/log"/>
<mkdir dir="${project.build.directory}/installer/bin"/>
</tasks>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>

8
skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/AlarmMessageProcessThread.java
View File

@ -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<String> acquireProcessedUsers() {
private List<String> 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);
}
}
}

398
skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/UserInfoCoordinator.java
View File

@ -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<String> 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<String> 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<String> users = AlarmMessageDao.selectAllUserIds();
// 查询当前有多少用户
List<String> users = AlarmMessageDao.selectAllUserIds();
logger.info("Query a total of {} process user", users.size());
// 将用户重新分配给服务
List<String> 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<String> 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<String> allocationUser(List<String> registeredThreads,
List<String> userIds) throws Exception {
List<String> realRedistributeThread = new ArrayList<String>();
Set<String> sortThreadIds = new HashSet<String>(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<String> allocationUser(List<String> registeredThreads,
List<String> userIds) {
List<String> realRedistributeThread = new ArrayList<String>();
Set<String> sortThreadIds = new HashSet<String>(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<String> acquireAllRegisteredThread() throws Exception {
return ZKUtil.getChildren(Config.ZKPath.REGISTER_SERVER_PATH);
}
}
return realRedistributeThread;
}
private boolean checkAllProcessStatus(List<String> registeredThreadIds,
ProcessThreadStatus status) throws Exception {
String registerPathPrefix = Config.ZKPath.REGISTER_SERVER_PATH + "/";
for (String threadId : registeredThreadIds) {
private List<String> acquireAllRegisteredThread() {
return ZKUtil.getChildren(Config.ZKPath.REGISTER_SERVER_PATH);
}
if (!ZKUtil.exists(Config.ZKPath.REGISTER_SERVER_PATH + "/"
+ threadId))
continue;
private boolean checkAllProcessStatus(List<String> 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<String> registeredThreadIds,
ProcessThreadStatus status) throws Exception {
Iterator<String> 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<String> 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);
}
}
}

3
skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/conf/Config.java
View File

@ -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 {

4
skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/dao/AlarmMessageDao.java
View File

@ -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"));

2
skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ProcessUtil.java
View File

@ -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);

27
skywalking-alarm/src/main/java/com/ai/cloud/skywalking/alarm/util/ZKUtil.java
View File

@ -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<String> 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<String>();
public static List<String> getChildren(String registerServerPath) throws Exception {
return client.getChildren().forPath(registerServerPath);
}
public static List<String> getChildrenWithWatcher(String registerServerPath, CuratorWatcher watcher) throws Exception {