My/bps_admin_java
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Improvement]通过引入多线程的方式大幅缩短了实例掉线检测功能的单次执行耗时

Browse Source
This commit is contained in:
mzaxd
2023-03-11 23:24:50 +08:00
parent b30725a3e2
commit 6538f32cbf
3 changed files with 414 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/mzaxd/noodles/job/CheckInstancesStatus.java
+56 -19
View File
@@ -10,6 +10,7 @@ import com.mzaxd.noodles.domain.entity.Container;
import com.mzaxd.noodles.domain.entity.HostDetector;
import com.mzaxd.noodles.domain.entity.HostMachine;
import com.mzaxd.noodles.service.*;
import com.mzaxd.noodles.util.CpuUtil;
import com.mzaxd.noodles.util.RedisCache;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
@@ -22,15 +23,17 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.*;
/**
* 检测所有实例的在线状态
*
* @author 13439
*/
@Slf4j
@Component
public class CheckInstancesStatus{
public class CheckInstancesStatus {
@Resource
private ContainerService containerService;
@@ -50,43 +53,77 @@ public class CheckInstancesStatus{
/**
* 检查实例状态并且发送对应的提醒
*/
@Scheduled(cron = "* 0/10 * * * ?")
@Scheduled(cron = "* 0/1 * * * ?")
public void checkInstancesStatus() {
//检测容器在线状态
List<Container> containers = containerService.list();
containers.forEach(container -> {
int thread = CpuUtil.getLogicProcessorCount();
int containerCount = containerService.count();
int corePoolSize = Math.min(thread + 1, containerCount);
int maxPoolSize = Math.max(thread + 1, containerCount);
// 创建一个包含10个线程的线程池
ExecutorService executor = new ThreadPoolExecutor(
corePoolSize,
maxPoolSize,
1,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
new ThreadPoolExecutor.AbortPolicy());
// 创建一个 Future 列表，用于存储每个容器检查的结果
List<Future<Container>> futures = new ArrayList<>();
for (Container container : containers) {
futures.add(executor.submit(() -> {
try {
if (!StringUtils.hasText(container.getWebUi())) {
container.setContainerState(SystemConstant.CONTAINER_STATE_UNKNOWN);
return;
return container;
}
HttpRequest.get(container.getWebUi()).setConnectionTimeout(1000).execute(true);
HttpRequest.get(container.getWebUi()).setConnectionTimeout(5000).execute(true);
log.info("[实例状态检测]：与{}建立连接成功", container.getName());
container.setContainerState(SystemConstant.CONTAINER_STATE_RUNNING);
} catch (Exception exception) {
log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", container.getName());
container.setContainerState(SystemConstant.CONTAINER_STATE_EXITED);
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
// 判断 Redis 里面有没有，如果有就不需要提醒，如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS);
if (!CollectionUtils.isEmpty(set)){
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (!CollectionUtils.isEmpty(set)) {
// 如果 Redis 里面有，说明已经发送过了未 check 的通知，所以不需要发送，直接返回
if (set.contains(container.getId().toString())) {
return;
return container;
} else {
//根据实例对应的提醒方式进行提醒
// 根据实例对应的提醒方式进行提醒
if (container.getNotify().equals(SystemConstant.NOTIFY_NO)) {
return;
return container;
} else {
notificationService.sendContainerOfflineNotification(container.getId());
}
}
}
//存入redis
// 存入 Redis
set.add(container.getId().toString());
redisCache.setCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS, set);
}
});
containerService.saveOrUpdateBatch(containers);
return container;
}));
}
// 等待所有线程执行完成，并收集更新后的容器列表
List<Container> updatedContainers = new ArrayList<>();
for (Future<Container> future : futures) {
try {
Container container = future.get();
updatedContainers.add(container);
} catch (InterruptedException | ExecutionException e) {
log.error("检查容器状态时出错：{}", e.getMessage());
}
}
// 将更新后的容器列表保存到数据库中
containerService.saveOrUpdateBatch(updatedContainers);
// 关闭线程池
executor.shutdown();
//检测虚拟机在线状态
LambdaQueryWrapper<HostMachine> vmWrapper = new LambdaQueryWrapper<>();
@@ -98,7 +135,7 @@ public class CheckInstancesStatus{
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_UNKNOWN);
return;
}
HttpRequest.get(vm.getManageIp()).setConnectionTimeout(1000).execute(true);
HttpRequest.get(vm.getManageIp()).setConnectionTimeout(5000).execute(true);
log.info("[实例状态检测]：与{}建立连接成功", vm.getName());
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE);
} catch (Exception exception) {
@@ -106,7 +143,7 @@ public class CheckInstancesStatus{
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_OFFLINE);
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_VM_IDS);
if (!CollectionUtils.isEmpty(set)){
if (!CollectionUtils.isEmpty(set)) {
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (set.contains(vm.getId().toString())) {
return;
@@ -132,7 +169,7 @@ public class CheckInstancesStatus{
List<HostMachine> hostMachines = new ArrayList<>();
hostDetectors.forEach(detector -> {
try {
HttpRequest.get(detector.getDetectorIpAddress() + UrlConstant.DETECTOR_IS_TRUE_URL).setConnectionTimeout(1000).execute(true);
HttpRequest.get(detector.getDetectorIpAddress() + UrlConstant.DETECTOR_IS_TRUE_URL).setConnectionTimeout(5000).execute(true);
HostMachine hostMachine = hostMachineService.getById(detector.getHostMachineId());
if (Objects.nonNull(hostMachine)) {
hostMachine.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE);
@@ -146,7 +183,7 @@ public class CheckInstancesStatus{
}
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_HOST_IDS);
if (!CollectionUtils.isEmpty(set)){
if (!CollectionUtils.isEmpty(set)) {
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (set.contains(hostMachine.getId().toString())) {
return;
src/main/java/com/mzaxd/noodles/util/CpuUtil.java
+19
View File
@@ -0,0 +1,19 @@
package com.mzaxd.noodles.util;
import oshi.SystemInfo;
import oshi.hardware.CentralProcessor;
import oshi.hardware.HardwareAbstractionLayer;
/**
* @author Mzaxd
* @since 2023-03-11 22:17
*/
public class CpuUtil {
public static int getLogicProcessorCount() {
SystemInfo systemInfo = new SystemInfo();
HardwareAbstractionLayer hardware = systemInfo.getHardware();
CentralProcessor processor = hardware.getProcessor();
return processor.getLogicalProcessorCount();
}
}
src/test/java/com/mzaxd/noodles/NoodlesApplicationTests.java
+323 -3
View File
@@ -8,10 +8,13 @@ import cn.hutool.http.HttpException;
import cn.hutool.http.HttpRequest;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.mzaxd.noodles.constant.RedisConstant;
import com.mzaxd.noodles.constant.SystemConstant;
import com.mzaxd.noodles.constant.UrlConstant;
import com.mzaxd.noodles.domain.entity.*;
import com.mzaxd.noodles.mapper.OsMapper;
import com.mzaxd.noodles.service.*;
import com.mzaxd.noodles.util.CpuUtil;
import com.mzaxd.noodles.util.RedisCache;
import com.mzaxd.noodles.util.SystemInfoUtils;
import lombok.extern.slf4j.Slf4j;
@@ -19,10 +22,12 @@ import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import javax.annotation.Resource;
import java.util.Map;
import java.util.Set;
import java.util.*;
import java.util.concurrent.*;
@Slf4j
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@@ -40,6 +45,12 @@ class NoodlesApplicationTests {
@Resource
private SystemSettingService systemSettingService;
@Resource
private NotificationService notificationService;
@Resource
private HostDetectorService hostDetectorService;
@Test
void contextLoads() {
try {
@@ -65,7 +76,7 @@ class NoodlesApplicationTests {
account.setFrom(systemSetting.getNotificationEmail());
account.setUser(systemSetting.getNotificationEmail());
account.setPass(systemSetting.getEmailPass());
MailUtil.send(account, CollUtil.newArrayList("mzaxd0712@gmail.com"),"测试5","邮件来自测试",false);
MailUtil.send(account, CollUtil.newArrayList("mzaxd0712@gmail.com"), "测试5", "邮件来自测试", false);
}
@Test
@@ -132,4 +143,313 @@ class NoodlesApplicationTests {
System.out.println(everyDayData);
}
@Test
public void checkInstanceStatus() {
long startTime = System.nanoTime(); // 记录开始时间
//检测容器在线状态
List<Container> containers = containerService.list();
containers.forEach(container -> {
try {
if (!StringUtils.hasText(container.getWebUi())) {
container.setContainerState(SystemConstant.CONTAINER_STATE_UNKNOWN);
return;
}
HttpRequest.get(container.getWebUi()).setConnectionTimeout(1000).execute(true);
log.info("[实例状态检测]：与{}建立连接成功", container.getName());
container.setContainerState(SystemConstant.CONTAINER_STATE_RUNNING);
} catch (Exception exception) {
log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", container.getName());
container.setContainerState(SystemConstant.CONTAINER_STATE_EXITED);
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS);
if (!CollectionUtils.isEmpty(set)) {
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (set.contains(container.getId().toString())) {
return;
} else {
//根据实例对应的提醒方式进行提醒
if (container.getNotify().equals(SystemConstant.NOTIFY_NO)) {
return;
} else {
notificationService.sendContainerOfflineNotification(container.getId());
}
}
}
//存入redis
set.add(container.getId().toString());
redisCache.setCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS, set);
}
});
containerService.saveOrUpdateBatch(containers);
//检测虚拟机在线状态
LambdaQueryWrapper<HostMachine> vmWrapper = new LambdaQueryWrapper<>();
vmWrapper.ne(HostMachine::getHostMachineId, SystemConstant.HOST_MACHINE_ID_HOST);
List<HostMachine> vms = hostMachineService.list(vmWrapper);
vms.forEach(vm -> {
try {
if (!StringUtils.hasText(vm.getManageIp())) {
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_UNKNOWN);
return;
}
HttpRequest.get(vm.getManageIp()).setConnectionTimeout(1000).execute(true);
log.info("[实例状态检测]：与{}建立连接成功", vm.getName());
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE);
} catch (Exception exception) {
log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", vm.getName());
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_OFFLINE);
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_VM_IDS);
if (!CollectionUtils.isEmpty(set)) {
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (set.contains(vm.getId().toString())) {
return;
} else {
//根据实例对应的提醒方式进行提醒
if (vm.getNotify().equals(SystemConstant.NOTIFY_NO)) {
return;
} else {
notificationService.sendVmOfflineNotification(vm.getId());
}
}
}
//存入redis
set.add(vm.getId().toString());
redisCache.setCacheSet(RedisConstant.NOTIFY_VM_IDS, set);
}
});
hostMachineService.saveOrUpdateBatch(vms);
//检测物理机在线状态(检测物理机要用探测器的isTureUrl接口)
LambdaQueryWrapper<HostDetector> detectorWrapper = new LambdaQueryWrapper<>();
List<HostDetector> hostDetectors = hostDetectorService.list(detectorWrapper);
List<HostMachine> hostMachines = new ArrayList<>();
hostDetectors.forEach(detector -> {
try {
HttpRequest.get(detector.getDetectorIpAddress() + UrlConstant.DETECTOR_IS_TRUE_URL).setConnectionTimeout(1000).execute(true);
HostMachine hostMachine = hostMachineService.getById(detector.getHostMachineId());
if (Objects.nonNull(hostMachine)) {
hostMachine.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE);
hostMachines.add(hostMachine);
}
} catch (Exception exception) {
HostMachine hostMachine = hostMachineService.getById(detector.getHostMachineId());
if (Objects.nonNull(hostMachine)) {
hostMachine.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_OFFLINE);
hostMachines.add(hostMachine);
}
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_HOST_IDS);
if (!CollectionUtils.isEmpty(set)) {
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (set.contains(hostMachine.getId().toString())) {
return;
} else {
//根据实例对应的提醒方式进行提醒
if (hostMachine.getNotify().equals(SystemConstant.NOTIFY_NO)) {
return;
} else {
notificationService.sendHostOfflineNotification(hostMachine.getId());
}
}
}
//存入redis
set.add(hostMachine.getId().toString());
redisCache.setCacheSet(RedisConstant.NOTIFY_HOST_IDS, set);
}
});
hostMachineService.saveOrUpdateBatch(hostMachines);
long endTime = System.nanoTime(); // 记录结束时间
long elapsedTime = endTime - startTime;
double seconds = (double) elapsedTime / 1_000_000_000.0; // 将纳秒转换为秒
System.out.println("代码执行时间：" + seconds + " 秒");
}
@Test
public void checkContainerStatusMultiThread() {
long startTime = System.nanoTime(); // 记录开始时间
List<Container> containers = containerService.list();
int thread = CpuUtil.getLogicProcessorCount();
int containerCount = containerService.count();
int corePoolSize = Math.min(thread + 1, containerCount);
int maxPoolSize = Math.max(thread + 1, containerCount);
// 创建一个包含10个线程的线程池
ExecutorService executor = new ThreadPoolExecutor(
corePoolSize,
maxPoolSize,
1,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
new ThreadPoolExecutor.AbortPolicy());
// 创建一个 Future 列表，用于存储每个容器检查的结果
List<Future<Container>> futures = new ArrayList<>();
for (Container container : containers) {
futures.add(executor.submit(() -> {
try {
if (!StringUtils.hasText(container.getWebUi())) {
container.setContainerState(SystemConstant.CONTAINER_STATE_UNKNOWN);
return container;
}
HttpRequest.get(container.getWebUi()).setConnectionTimeout(5000).execute(true);
log.info("[实例状态检测]：与{}建立连接成功", container.getName());
container.setContainerState(SystemConstant.CONTAINER_STATE_RUNNING);
} catch (Exception exception) {
log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", container.getName());
container.setContainerState(SystemConstant.CONTAINER_STATE_EXITED);
// 判断 Redis 里面有没有，如果有就不需要提醒，如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS);
if (!CollectionUtils.isEmpty(set)) {
// 如果 Redis 里面有，说明已经发送过了未 check 的通知，所以不需要发送，直接返回
if (set.contains(container.getId().toString())) {
return container;
} else {
// 根据实例对应的提醒方式进行提醒
if (container.getNotify().equals(SystemConstant.NOTIFY_NO)) {
return container;
} else {
notificationService.sendContainerOfflineNotification(container.getId());
}
}
}
// 存入 Redis
set.add(container.getId().toString());
redisCache.setCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS, set);
}
return container;
}));
}
// 等待所有线程执行完成，并收集更新后的容器列表
List<Container> updatedContainers = new ArrayList<>();
for (Future<Container> future : futures) {
try {
Container container = future.get();
updatedContainers.add(container);
} catch (InterruptedException | ExecutionException e) {
log.error("检查容器状态时出错：{}", e.getMessage());
}
}
// 将更新后的容器列表保存到数据库中
containerService.saveOrUpdateBatch(updatedContainers);
// 关闭线程池
executor.shutdown();
long endTime = System.nanoTime(); // 记录结束时间
long elapsedTime = endTime - startTime;
double seconds = (double) elapsedTime / 1_000_000_000.0; // 将纳秒转换为秒
System.out.println("代码执行时间：" + seconds + " 秒");
}
@Test
public void checkVmStatus() {
long startTime = System.nanoTime(); // 记录开始时间
//检测虚拟机在线状态
LambdaQueryWrapper<HostMachine> vmWrapper = new LambdaQueryWrapper<>();
vmWrapper.ne(HostMachine::getHostMachineId, SystemConstant.HOST_MACHINE_ID_HOST);
List<HostMachine> vms = hostMachineService.list(vmWrapper);
vms.forEach(vm -> {
try {
if (!StringUtils.hasText(vm.getManageIp())) {
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_UNKNOWN);
return;
}
HttpRequest.get(vm.getManageIp()).setConnectionTimeout(5000).execute(true);
log.info("[实例状态检测]：与{}建立连接成功", vm.getName());
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE);
} catch (Exception exception) {
log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", vm.getName());
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_OFFLINE);
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_VM_IDS);
if (!CollectionUtils.isEmpty(set)) {
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (set.contains(vm.getId().toString())) {
return;
} else {
//根据实例对应的提醒方式进行提醒
if (vm.getNotify().equals(SystemConstant.NOTIFY_NO)) {
return;
} else {
notificationService.sendVmOfflineNotification(vm.getId());
}
}
}
//存入redis
set.add(vm.getId().toString());
redisCache.setCacheSet(RedisConstant.NOTIFY_VM_IDS, set);
}
});
hostMachineService.saveOrUpdateBatch(vms);
long endTime = System.nanoTime(); // 记录结束时间
long elapsedTime = endTime - startTime;
double seconds = (double) elapsedTime / 1_000_000_000.0; // 将纳秒转换为秒
System.out.println("代码执行时间：" + seconds + " 秒");
}
@Test
public void checkVmStatusMultiThread() throws InterruptedException {
long startTime = System.nanoTime(); // 记录开始时间
ExecutorService executorService = Executors.newFixedThreadPool(10); // 创建一个线程池
List<HostMachine> vms = hostMachineService.list(); // 获取所有主机列表
CountDownLatch countDownLatch = new CountDownLatch(vms.size()); // 用于等待所有线程完成
for (HostMachine vm : vms) {
executorService.submit(() -> {
try {
if (!StringUtils.hasText(vm.getManageIp())) {
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_UNKNOWN);
return;
}
HttpRequest.get(vm.getManageIp()).setConnectionTimeout(5000).execute(true);
log.info("[实例状态检测]：与{}建立连接成功", vm.getName());
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE);
} catch (Exception exception) {
log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", vm.getName());
vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_OFFLINE);
//判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒
Set<String> set = redisCache.getCacheSet(RedisConstant.NOTIFY_VM_IDS);
if (!CollectionUtils.isEmpty(set)) {
//如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回
if (set.contains(vm.getId().toString())) {
return;
} else {
//根据实例对应的提醒方式进行提醒
if (vm.getNotify().equals(SystemConstant.NOTIFY_NO)) {
return;
} else {
notificationService.sendVmOfflineNotification(vm.getId());
}
}
}
//存入redis
set.add(vm.getId().toString());
redisCache.setCacheSet(RedisConstant.NOTIFY_VM_IDS, set);
} finally {
countDownLatch.countDown(); // 完成一个线程
}
});
}
countDownLatch.await(); // 等待所有线程完成
hostMachineService.saveOrUpdateBatch(vms); // 保存更新后的主机状态
executorService.shutdown(); // 关闭线程池
long endTime = System.nanoTime(); // 记录结束时间
long elapsedTime = endTime - startTime;
double seconds = (double) elapsedTime / 1_000_000_000.0; // 将纳秒转换为秒
System.out.println("代码执行时间：" + seconds + " 秒");
}
}