|
...
|
...
|
@@ -63,10 +63,10 @@ public class ElectionService { |
|
|
|
|
|
|
|
Container.CONTAINER.putAllClient(jobService.getAllClient());
|
|
|
|
// 监听客户端
|
|
|
|
// listenClientNode(jobsInfo);
|
|
|
|
// listenClientNode(jobsInfo);
|
|
|
|
listenClient();
|
|
|
|
// 调度开始,只有在触发调度的时候再选取执行节点
|
|
|
|
jobScheduleService.batchSchedule(Container.CONTAINER.getAllJobs());
|
|
|
|
// jobScheduleService.batchSchedule(Container.CONTAINER.getAllJobs());
|
|
|
|
|
|
|
|
count.await();
|
|
|
|
}
|
|
...
|
...
|
@@ -91,122 +91,6 @@ public class ElectionService { |
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
private void listenClientNode(Map<String, List<JobInfo>> jobsInfo) throws Exception {
|
|
|
|
for (Map.Entry<String, List<JobInfo>> entry : jobsInfo.entrySet()) {
|
|
|
|
buildWatch(entry.getKey());
|
|
|
|
watchJobInfo(entry.getKey());
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
private void buildWatch(String module) throws Exception {
|
|
|
|
TreeCache treeCache = new TreeCache(client, CLIENT_NODES + module + "/clientnode");
|
|
|
|
treeCache.getListenable().addListener(new TreeCacheListener() {
|
|
|
|
@Override
|
|
|
|
public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
|
|
|
|
ChildData data = event.getData();
|
|
|
|
if (data != null) {
|
|
|
|
String path = data.getPath();
|
|
|
|
String clientIp = path.substring(path.lastIndexOf("/") + 1);
|
|
|
|
switch (event.getType()) {
|
|
|
|
case NODE_ADDED:
|
|
|
|
logger.info("register new client, the client is {}", String.valueOf(data.getData()));
|
|
|
|
Container.CONTAINER.addClient(module, clientIp);
|
|
|
|
break;
|
|
|
|
case NODE_REMOVED:
|
|
|
|
// 有客户端下线，需要将下线的客户端上执行的任务切换到其他节点上
|
|
|
|
logger.info("remove node {}", data.getPath());
|
|
|
|
// String path = data.getPath();
|
|
|
|
// String clientIp = path.substring(path.lastIndexOf("/") + 1);
|
|
|
|
Container.CONTAINER.removeClient(module, clientIp);
|
|
|
|
// 重新选取节点
|
|
|
|
break;
|
|
|
|
case NODE_UPDATED:
|
|
|
|
break;
|
|
|
|
|
|
|
|
default:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
logger.info("data is null : " + event.getType());
|
|
|
|
}
|
|
|
|
}
|
|
|
|
});
|
|
|
|
// 开始监听
|
|
|
|
treeCache.start();
|
|
|
|
}
|
|
|
|
|
|
|
|
private void watchJobInfo(String module) throws Exception {
|
|
|
|
TreeCache treeCache = new TreeCache(client, CLIENT_NODES + module + "/jobInfo");
|
|
|
|
treeCache.getListenable().addListener(new TreeCacheListener() {
|
|
|
|
@Override
|
|
|
|
public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
|
|
|
|
ChildData data = event.getData();
|
|
|
|
if (data != null) {
|
|
|
|
JobInfo jobInfo = JSON.parseObject(data.getData(), JobInfo.class);
|
|
|
|
switch (event.getType()) {
|
|
|
|
// TODO 若先启动服务端，后启动客户端，会不会有问题---有问题
|
|
|
|
case NODE_ADDED:
|
|
|
|
logger.info("create new Job, the Job is {}", String.valueOf(data.getData()));
|
|
|
|
ScheduleJobInfo scheduleJobInfo = new ScheduleJobInfo();
|
|
|
|
scheduleJobInfo.setJobInfo(jobInfo);
|
|
|
|
scheduleJobInfo.setFuture(null);
|
|
|
|
Container.CONTAINER.addJobInfo(jobInfo.getModule(), scheduleJobInfo);
|
|
|
|
// if (jobInfo.getOperator() == ConstantEnum.JobOperatorEnum.EXEC.getValue()) {
|
|
|
|
jobInfo.setOperator(ConstantEnum.JobOperatorEnum.TRIGGER.getValue());
|
|
|
|
jobInfo.setStatus(ConstantEnum.JobStatusEnum.STARTING.getValue());
|
|
|
|
jobScheduleService.schedule(scheduleJobInfo);
|
|
|
|
// }
|
|
|
|
break;
|
|
|
|
case NODE_REMOVED:
|
|
|
|
|
|
|
|
break;
|
|
|
|
case NODE_UPDATED:
|
|
|
|
logger.info("update Job, the Job is {}", String.valueOf(data.getData()));
|
|
|
|
// 如果是任务的启动则需要schedule
|
|
|
|
ScheduleJobInfo updateScheduleJobInfo = Container.CONTAINER.getJobsByModule(jobInfo.getModule())
|
|
|
|
.get(jobInfo.getJobName());
|
|
|
|
if (jobInfo.getOperator() == ConstantEnum.JobOperatorEnum.EXEC.getValue()) {
|
|
|
|
jobInfo.setOperator(ConstantEnum.JobOperatorEnum.TRIGGER.getValue());
|
|
|
|
jobInfo.setStatus(ConstantEnum.JobStatusEnum.STARTING.getValue());
|
|
|
|
|
|
|
|
updateScheduleJobInfo.setJobInfo(jobInfo);
|
|
|
|
updateScheduleJobInfo.setFuture(null);
|
|
|
|
jobScheduleService.schedule(updateScheduleJobInfo);
|
|
|
|
} else if (jobInfo.getOperator() == ConstantEnum.JobOperatorEnum.STOP.getValue()) {
|
|
|
|
// 当前任务是执行状态时，才可以停止，然后cancel之
|
|
|
|
if (jobInfo.getStatus() == ConstantEnum.JobStatusEnum.STARTING.getValue()) {
|
|
|
|
jobInfo.setStatus(ConstantEnum.JobStatusEnum.STOPPING.getValue());
|
|
|
|
updateScheduleJobInfo.setJobInfo(jobInfo);
|
|
|
|
updateScheduleJobInfo.getFuture().cancel(false);
|
|
|
|
// 重新更新到zk上
|
|
|
|
String jobPath = CLIENT_NODES + jobInfo.getModule() + "/jobinfo/"
|
|
|
|
+ jobInfo.getJobName();
|
|
|
|
try {
|
|
|
|
client.setData().forPath(jobPath,
|
|
|
|
JSON.toJSONString(jobInfo).getBytes(Charset.forName("UTF-8")));
|
|
|
|
} catch (Exception e) {
|
|
|
|
logger.error("stop job faild , job is {}, exception {}", jobInfo, e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
} else if (jobInfo.getOperator() == ConstantEnum.JobOperatorEnum.UPDATE.getValue()) {
|
|
|
|
updateScheduleJobInfo.setJobInfo(jobInfo);
|
|
|
|
}
|
|
|
|
|
|
|
|
break;
|
|
|
|
|
|
|
|
default:
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
logger.info("data is null : " + event.getType());
|
|
|
|
}
|
|
|
|
}
|
|
|
|
});
|
|
|
|
// 开始监听
|
|
|
|
treeCache.start();
|
|
|
|
}
|
|
|
|
|
|
|
|
private void buildScheduleJobInfoContainer(Map<String, List<JobInfo>> jobsInfo) {
|
|
|
|
for (Map.Entry<String, List<JobInfo>> entry : jobsInfo.entrySet()) {
|
|
|
|
String moduleName = entry.getKey();
|
|
...
|
...
|
@@ -250,17 +134,18 @@ public class ElectionService { |
|
|
|
scheduleJobInfo.setJobInfo(jobInfo);
|
|
|
|
scheduleJobInfo.setFuture(null);
|
|
|
|
Container.CONTAINER.addJobInfo(jobInfo.getModule(), scheduleJobInfo);
|
|
|
|
jobInfo.setOperator(ConstantEnum.JobOperatorEnum.TRIGGER.getValue());
|
|
|
|
jobInfo.setStatus(ConstantEnum.JobStatusEnum.STARTING.getValue());
|
|
|
|
jobScheduleService.schedule(scheduleJobInfo);
|
|
|
|
if (jobInfo.getOperator() == ConstantEnum.JobOperatorEnum.EXEC.getValue()
|
|
|
|
|| jobInfo.getOperator() == ConstantEnum.JobOperatorEnum.TRIGGER.getValue()) {
|
|
|
|
jobInfo.setOperator(ConstantEnum.JobOperatorEnum.TRIGGER.getValue());
|
|
|
|
jobInfo.setStatus(ConstantEnum.JobStatusEnum.STARTING.getValue());
|
|
|
|
jobScheduleService.schedule(scheduleJobInfo);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
break;
|
|
|
|
case NODE_REMOVED:
|
|
|
|
if (isClientInfoType(pathType)) {
|
|
|
|
// 有客户端下线，需要将下线的客户端上执行的任务切换到其他节点上
|
|
|
|
logger.info("remove node {}", data.getPath());
|
|
|
|
// String path = data.getPath();
|
|
|
|
// String clientIp = path.substring(path.lastIndexOf("/") + 1);
|
|
|
|
Container.CONTAINER.removeClient(parseModule(path),
|
|
|
|
path.substring(path.lastIndexOf("/") + 1));
|
|
|
|
// 重新选取节点
|
|
...
|
...
|
@@ -289,7 +174,7 @@ public class ElectionService { |
|
|
|
updateScheduleJobInfo.setJobInfo(jobInfo);
|
|
|
|
updateScheduleJobInfo.getFuture().cancel(false);
|
|
|
|
// 重新更新到zk上
|
|
|
|
String jobPath = CLIENT_NODES + jobInfo.getModule() + "/jobinfo/"
|
|
|
|
String jobPath = CLIENT_NODES + "/" + jobInfo.getModule() + "/jobinfo/"
|
|
|
|
+ jobInfo.getJobName();
|
|
|
|
try {
|
|
|
|
client.setData().forPath(jobPath,
|
|
...
|
...
|
@@ -332,7 +217,7 @@ public class ElectionService { |
|
|
|
}
|
|
|
|
|
|
|
|
private String parseModule(String path) {
|
|
|
|
String s = path.substring(CLIENT_NODES.length()+1);
|
|
|
|
String s = path.substring(CLIENT_NODES.length() + 1);
|
|
|
|
String module = s.substring(0, s.indexOf("/"));
|
|
|
|
return module;
|
|
|
|
}
|
...
|
...
|
|
