From f2d313721f9afacefc811c06c11249f009a5cdc6 Mon Sep 17 00:00:00 2001 From: Mzaxd Date: Thu, 23 Mar 2023 09:14:28 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=96=B0=E9=83=A8=E7=BD=B2=E4=B8=80?= =?UTF-8?q?=E4=B8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../noodles/job/CheckInstancesStatus.java | 2 +- .../noodles/NoodlesApplicationTests.java | 183 +++++++++++++++--- 2 files changed, 153 insertions(+), 32 deletions(-) diff --git a/src/main/java/com/mzaxd/noodles/job/CheckInstancesStatus.java b/src/main/java/com/mzaxd/noodles/job/CheckInstancesStatus.java index 5c17293..869f9aa 100644 --- a/src/main/java/com/mzaxd/noodles/job/CheckInstancesStatus.java +++ b/src/main/java/com/mzaxd/noodles/job/CheckInstancesStatus.java @@ -61,7 +61,7 @@ public class CheckInstancesStatus { int corePoolSize = Math.min(thread + 1, containerCount); int maxPoolSize = Math.max(thread + 1, containerCount); - // 创建一个包含10个线程的线程池 + // 创建一个线程池 ExecutorService executor = new ThreadPoolExecutor( corePoolSize, maxPoolSize, diff --git a/src/test/java/com/mzaxd/noodles/NoodlesApplicationTests.java b/src/test/java/com/mzaxd/noodles/NoodlesApplicationTests.java index eb3ee92..e092401 100644 --- a/src/test/java/com/mzaxd/noodles/NoodlesApplicationTests.java +++ b/src/test/java/com/mzaxd/noodles/NoodlesApplicationTests.java @@ -412,48 +412,169 @@ class NoodlesApplicationTests { public void checkVmStatusMultiThread() throws InterruptedException { long startTime = System.nanoTime(); // 记录开始时间 - ExecutorService executorService = Executors.newFixedThreadPool(10); // 创建一个线程池 - List vms = hostMachineService.list(); // 获取所有主机列表 - CountDownLatch countDownLatch = new CountDownLatch(vms.size()); // 用于等待所有线程完成 - for (HostMachine vm : vms) { - executorService.submit(() -> { + List 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); + + ExecutorService executor = new ThreadPoolExecutor( + corePoolSize, + maxPoolSize, + 1, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(), + new ThreadPoolExecutor.AbortPolicy()); + + // 创建一个 Future 列表，用于存储每个容器检查的结果 + List> futures = new ArrayList<>(); + + for (Container container : containers) { + futures.add(executor.submit(() -> { try { - if (!StringUtils.hasText(vm.getManageIp())) { - vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_UNKNOWN); - return; + if (!StringUtils.hasText(container.getWebUi()) && !StringUtils.hasText(container.getServerAddress())) { + container.setContainerState(SystemConstant.CONTAINER_STATE_UNKNOWN); + return container; } - HttpRequest.get(vm.getManageIp()).setConnectionTimeout(5000).execute(true); - log.info("[实例状态检测]：与{}建立连接成功", vm.getName()); - vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE); + //通过Socket（IP+端口）判断是否在线 + if (StringUtils.hasText(container.getServerAddress())) { + if (UrlUtil.isServiceOnline(UrlUtil.getHostname(container.getServerAddress()), UrlUtil.getPort(container.getServerAddress()))) { + container.setContainerState(SystemConstant.CONTAINER_STATE_RUNNING); + return container; + } + } + //通过HTTP请求判断是否在线 + HttpRequest.get(container.getWebUi()).setConnectionTimeout(5000).execute(true); + log.info("[实例状态检测]：与{}建立连接成功", container.getName()); + container.setContainerState(SystemConstant.CONTAINER_STATE_RUNNING); } catch (Exception exception) { - log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", vm.getName()); - vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_OFFLINE); - //判断Redis里面有没有 如果有就不需要提醒 如果没有就提醒 - Set set = redisCache.getCacheSet(RedisConstant.NOTIFY_VM_IDS); + log.info("[实例状态检测]：与{}建立连接失败，状态转为离线", container.getName()); + container.setContainerState(SystemConstant.CONTAINER_STATE_EXITED); + // 判断 Redis 里面有没有，如果有就不需要提醒，如果没有就提醒 + Set set = redisCache.getCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS); if (!CollectionUtils.isEmpty(set)) { - //如果redis里面有 说明已经发送过了未check的通知 所以不需要发送 直接返回 - if (set.contains(vm.getId().toString())) { - return; + // 如果 Redis 里面有，说明已经发送过了未 check 的通知，所以不需要发送，直接返回 + if (set.contains(container.getId().toString())) { + return container; } else { - //根据实例对应的提醒方式进行提醒 - if (vm.getNotify().equals(SystemConstant.NOTIFY_NO)) { - return; + // 根据实例对应的提醒方式进行提醒 + if (container.getNotify().equals(SystemConstant.NOTIFY_NO)) { + return container; } else { - notificationService.sendVmOfflineNotification(vm.getId()); + notificationService.sendContainerOfflineNotification(container.getId()); } } } - //存入redis - set.add(vm.getId().toString()); - redisCache.setCacheSet(RedisConstant.NOTIFY_VM_IDS, set); - } finally { - countDownLatch.countDown(); // 完成一个线程 + // 存入 Redis + set.add(container.getId().toString()); + redisCache.setCacheSet(RedisConstant.NOTIFY_CONTAINER_IDS, set); } - }); + return container; + })); } - countDownLatch.await(); // 等待所有线程完成 - hostMachineService.saveOrUpdateBatch(vms); // 保存更新后的主机状态 - executorService.shutdown(); // 关闭线程池 + + // 等待所有线程执行完成，并收集更新后的容器列表 + List updatedContainers = new ArrayList<>(); + for (Future future : futures) { + try { + Container container = future.get(); + updatedContainers.add(container); + } catch (InterruptedException | ExecutionException e) { + log.error("检查容器状态时出错：{}", e.getMessage()); + } + } + + // 将更新后的容器列表保存到数据库中 + containerService.saveOrUpdateBatch(updatedContainers); + // 关闭线程池 + executor.shutdown(); + + //检测虚拟机在线状态 + LambdaQueryWrapper vmWrapper = new LambdaQueryWrapper<>(); + vmWrapper.ne(HostMachine::getHostMachineId, SystemConstant.HOST_MACHINE_ID_HOST); + List vms = hostMachineService.list(vmWrapper); + vms.forEach(vm -> { + try { + if (!StringUtils.hasText(vm.getServerAddress()) && !StringUtils.hasText(vm.getServerAddress())) { + vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_UNKNOWN); + return; + } + //通过Ping的方式判断是否在线 + if (StringUtils.hasText(vm.getServerAddress())) { + if (UrlUtil.isHostOnline(vm.getServerAddress())){ + log.info("[实例状态检测]：与{}建立连接成功", vm.getName()); + vm.setHostMachineState(SystemConstant.HOST_MACHINE_STATE_ONLINE); + return; + } + } + //通过HTTP请求的方式判断是否在线 + 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 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 detectorWrapper = new LambdaQueryWrapper<>(); + List hostDetectors = hostDetectorService.list(detectorWrapper); + List hostMachines = new ArrayList<>(); + hostDetectors.forEach(detector -> { + try { + 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); + 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 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;