potimize threadPool and thread idling

PaasMonitor-client/src/main/java/com/nju/edu/paasMonitorClient/MonitorClient.java

@@ -107,16 +107,17 @@ public class MonitorClient {
 ////            logError(SeecSystem.BOK, "try1", "t", "error", e);
 ////            logError(SeecSystem.BOK, "try2", "t1", "error", e);
 ////            logError(SeecSystem.BOK, "try1", "t", "error", e);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.FAIL);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.FAIL);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
-            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.FAIL);
+//            Thread.sleep(3000);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.FAIL);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
+//            logEvent(SeecSystem.PAAS, "tryEvent", "tryEvent", EventStatus.SUCCESS);
 //        }
     }
 }

PaasMonitor-client/src/main/java/com/nju/edu/paasMonitorClient/consumer/MessageManager.java

@@ -1,41 +0,0 @@
-package com.nju.edu.paasMonitorClient.consumer;
-
-import com.nju.edu.paasMonitorClient.entity.Message;
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
-@Slf4j
-public class MessageManager {
-    private static volatile MessageManager instance;
-    private final BlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>();
-
-    private MessageManager() {
-        init();
-    }
-
-    public static MessageManager getInstance() {
-        if (instance == null) {
-            synchronized (MessageManager.class) {
-                if (instance == null) {
-                    instance = new MessageManager();
-                }
-            }
-        }
-        return instance;
-    }
-
-    public void add(Message message) {
-        boolean success = messageQueue.offer(message);
-        if (!success) {
-            log.error("Can't add message to queue");
-        }
-    }
-
-    private void init() {
-        MonitorConsumer monitorConsumer = new MonitorConsumer(messageQueue);
-        Thread thread = new Thread(monitorConsumer);
-        thread.start();
-    }
-}

PaasMonitor-client/src/main/java/com/nju/edu/paasMonitorClient/consumer/MonitorConsumer.java

@@ -4,6 +4,7 @@ import com.nju.edu.paasMonitorClient.config.Config;
 import com.nju.edu.paasMonitorClient.entity.*;
 import com.nju.edu.paasMonitorClient.entity.Error;
 import com.nju.edu.paasMonitorClient.util.PostHelper;
+import com.nju.edu.paasMonitorClient.util.SystemInfoUtil;
 import com.nju.edu.paasMonitorCommon.dto.*;
 import lombok.extern.slf4j.Slf4j;
 
@@ -12,25 +13,52 @@ import java.util.concurrent.*;
 
 @Slf4j
 public class MonitorConsumer implements Runnable {
-    private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(4, 8,
-            10000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(40));
-    private int errorCount = 0;
-    private BlockingQueue<Message> messageQueue;
+    private static final int CORE_POOL_SIZE = SystemInfoUtil.getCpuCoreSize() * 2;
+    private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE,
+            CORE_POOL_SIZE, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
+    private int threadPoolErrorCount = 0;
+    private int mqErrorCount = 0;
+    private final BlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>(2000);
     private final String serverKey = "server.address.prefix";
-    private static final int REJECT_SLEEP_TIME = 50;
+    private volatile static MonitorConsumer instance;
+    private boolean active = true;
 
-    public MonitorConsumer(BlockingQueue<Message> queue) {
-        this.messageQueue = queue;
+    public static MonitorConsumer getInstance() {
+        if (instance == null || !instance.active) {
+            synchronized (MonitorConsumer.class) {
+                if (instance == null || !instance.active) {
+                    instance = new MonitorConsumer();
+                    Thread thread = new Thread(instance);
+                    thread.start();
+                }
+            }
+        }
+        return instance;
+    }
+
+    public void add(Message message) {
+        boolean success = messageQueue.offer(message);
+        if (!success) {
+            mqErrorCount++;
+            if (mqErrorCount == 1 || mqErrorCount % 50 == 0) {
+                log.error("Can't add message to queue");
+            }
+        }
+    }
+
+    private MonitorConsumer() {
     }
 
     @Override
     public void run() {
-        while (true) {
+        while (active || !messageQueue.isEmpty()) {
             Message backup = null;
             try {
                 Message message = messageQueue.poll(10, TimeUnit.MILLISECONDS);
                 backup = message;
-                if (message != null) {
+                if (message == null) {
+                    active = false;
+                } else {
                     threadPoolExecutor.execute(() -> {
                         MessageDto send;
                         try {
@@ -38,21 +66,21 @@ public class MonitorConsumer implements Runnable {
                             if (message instanceof Call) {
                                 address += "call";
                                 send = new CallDto(message.getSystem(), message.getCategory(),
-                                        message.getName(),message.getCreateTime(),((Call) message).getDuration(),
+                                        message.getName(), message.getCreateTime(), ((Call) message).getDuration(),
                                         ((Call) message).getStatus().getVal());
                             } else if (message instanceof Error) {
                                 address += "error";
                                 send = new ErrorDto(message.getSystem(), message.getCategory(),
-                                        message.getName(),message.getCreateTime(),((Error) message).getMessage(),
+                                        message.getName(), message.getCreateTime(), ((Error) message).getMessage(),
                                         ((Error) message).getExceptionInfo(), ((Error) message).getStackTrace());
                             } else if (message instanceof Event) {
                                 address += "event";
                                 send = new EventDto(message.getSystem(), message.getCategory(),
-                                        message.getName(),message.getCreateTime(),((Event) message).getStatus().getVal());
+                                        message.getName(), message.getCreateTime(), ((Event) message).getStatus().getVal());
                             } else if (message instanceof Metric) {
                                 address += "metric";
                                 send = new MetricDto(message.getSystem(), message.getCategory(),
-                                        message.getName(),message.getCreateTime(),((Metric) message).getCount(),
+                                        message.getName(), message.getCreateTime(), ((Metric) message).getCount(),
                                         ((Metric) message).getType().getVal());
                             } else {
                                 throw new Exception("Wrong type of message");
@@ -67,21 +95,18 @@ public class MonitorConsumer implements Runnable {
                 }
             } catch (InterruptedException ignored) {
             } catch (RejectedExecutionException e) {
-                if(backup != null){
+                if (backup != null) {
                     boolean success = messageQueue.offer(backup);
-                    if(!success){
-                        log.error("Can't add message to queue");
+                    if (!success) {
+                        log.error("Can't add message to queue", e);
                     }
                 }
-                errorCount++;
-                if (errorCount == 1 || errorCount % 50 == 0) {
-                    log.error("ThreadPool is full", e);
-                }
-                try {
-                    Thread.sleep(REJECT_SLEEP_TIME);
-                }catch (InterruptedException ignore){
+                threadPoolErrorCount++;
+                if (threadPoolErrorCount == 1 || threadPoolErrorCount % 50 == 0) {
+                    log.error("ThreadPool is full");
                 }
             }
         }
+        threadPoolExecutor.shutdown();
     }
 }

PaasMonitor-client/src/main/java/com/nju/edu/paasMonitorClient/entity/Message.java

@@ -1,6 +1,6 @@
 package com.nju.edu.paasMonitorClient.entity;
 
-import com.nju.edu.paasMonitorClient.consumer.MessageManager;
+import com.nju.edu.paasMonitorClient.consumer.MonitorConsumer;
 import lombok.Getter;
 import lombok.ToString;
 
@@ -15,6 +15,6 @@ public class Message {
     public Message(){}
 
     public void complete(){
-        MessageManager.getInstance().add(this);
+        MonitorConsumer.getInstance().add(this);
     }
 }

PaasMonitor-client/src/main/java/com/nju/edu/paasMonitorClient/util/SystemInfoUtil.java

@@ -0,0 +1,7 @@
+package com.nju.edu.paasMonitorClient.util;
+
+public class SystemInfoUtil {
+    public static int getCpuCoreSize(){
+        return Runtime.getRuntime().availableProcessors();
+    }
+}

PaasMonitor-client/target/maven-status/maven-compiler-plugin/compile/default-compile/createdFiles.lst

@@ -3,8 +3,8 @@ com\nju\edu\paasMonitorClient\entity\Event.class
 com\nju\edu\paasMonitorClient\MonitorClient.class
 com\nju\edu\paasMonitorClient\config\Config.class
 com\nju\edu\paasMonitorClient\entity\Error.class
-com\nju\edu\paasMonitorClient\consumer\MessageManager.class
 com\nju\edu\paasMonitorClient\entity\Message.class
+com\nju\edu\paasMonitorClient\util\SystemInfoUtil.class
 com\nju\edu\paasMonitorClient\entity\Call.class
 com\nju\edu\paasMonitorClient\entity\Metric.class
 com\nju\edu\paasMonitorClient\consumer\MonitorConsumer.class

PaasMonitor-client/target/maven-status/maven-compiler-plugin/compile/default-compile/inputFiles.lst

@@ -1,5 +1,4 @@
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\consumer\MonitorConsumer.java
-D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\consumer\MessageManager.java
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\entity\Event.java
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\MonitorClient.java
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\util\PostHelper.java
@@ -7,4 +6,5 @@ D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClie
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\entity\Metric.java
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\entity\Call.java
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\entity\Error.java
+D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\util\SystemInfoUtil.java
 D:\SEEC\PaasMonitor\PaasMonitor-client\src\main\java\com\nju\edu\paasMonitorClient\entity\Message.java

PaasMonitor-server/src/main/java/com/nju/edu/paasMonitorServer/alert/process/sender/AlertSender.java

@@ -29,17 +29,18 @@ public class AlertSender implements Runnable {
     private volatile static AlertSender instance;
     private final BlockingQueue<AlertMessage> mq = new LinkedBlockingQueue<>();
     private int errorCount = 0;
+    private boolean active = true;
 
     private AlertSender() {
-        Thread thread = new Thread(this);
-        thread.start();
     }
 
     public static AlertSender getInstance() {
-        if (instance == null) {
+        if (instance == null || !instance.active) {
             synchronized (AlertSender.class) {
-                if (instance == null) {
+                if (instance == null || !instance.active) {
                     instance = new AlertSender();
+                    Thread thread = new Thread(instance);
+                    thread.start();
                 }
             }
         }
@@ -48,10 +49,12 @@ public class AlertSender implements Runnable {
 
     @Override
     public void run() {
-        while (true) {
+        while (active || !mq.isEmpty()) {
             try {
-                AlertMessage alertMessage = mq.poll(5, TimeUnit.MILLISECONDS);
-                if (alertMessage != null) {
+                AlertMessage alertMessage = mq.poll(10, TimeUnit.MILLISECONDS);
+                if (alertMessage == null) {
+                    active = false;
+                } else {
                     Set<String> receivers = filterReceivers(configService.getAlertReceiver(), alertMessage.getConfigPO());
                     configService.saveAlertMessage(alertMessage);
                     mailSendService.sendMail(receivers, alertMessage.buildSubject(), alertMessage.buildText());

PaasMonitor-server/src/main/java/com/nju/edu/paasMonitorServer/alert/process/sender/alertmessage/AlertList.java

@@ -23,11 +23,17 @@ public class AlertList extends AlertMessage {
         for (SubConditionPO subConditionPO : map.keySet()) {
             if (subConditionPO.getType() == SubConditionType.AVG_MIN.getVal()
                     || subConditionPO.getType() == SubConditionType.AVG_MAX.getVal()) {
-                text.append(String.format("[%s阈值:%.2f, 实际值: %.2f] ",
+                text.append(String.format("[%s阈值:%.3f, 实际值: %.3f] ",
                         SubConditionType.getDescriptionByVal(subConditionPO.getType()),
                         subConditionPO.getValue(), map.get(subConditionPO)));
+            } else if (subConditionPO.getType() == SubConditionType.MAX_RISE_VOLATILITY_PERCENT.getVal()
+                    || subConditionPO.getType() == SubConditionType.MAX_FALL_VOLATILITY_PERCENT.getVal()) {
+                text.append(String.format("[%s阈值:%.2f%%, 限制频次:%d, 实际频次:%d] ",
+                        SubConditionType.getDescriptionByVal(subConditionPO.getType()),
+                        subConditionPO.getValue().doubleValue() * 100, subConditionPO.getFrequency(),
+                        map.get(subConditionPO).intValue()));
             } else {
-                text.append(String.format("[%s阈值:%.2f, 限制频次:%d, 实际频次:%d] ",
+                text.append(String.format("[%s阈值:%.3f, 限制频次:%d, 实际频次:%d] ",
                         SubConditionType.getDescriptionByVal(subConditionPO.getType()),
                         subConditionPO.getValue(), subConditionPO.getFrequency(),
                         map.get(subConditionPO).intValue()));

PaasMonitor-server/src/main/java/com/nju/edu/paasMonitorServer/alert/process/sender/alertmessage/AlertMessage.java

@@ -30,11 +30,13 @@ public class AlertMessage {
         int alertLevel = this.configPO.getAlertLevel();
         return String.format("[Paas Monitor %s告警] : [%s - %s - %s] [级别: %s]",
                 AlertType.getDescriptionByVal(type), SeecSystem.getDescriptionByVal(systemId),
-                configPO.getCategory(), configPO.getName(), AlertLevel.getDescriptionByVal(alertLevel));
+                configPO.getCategory(), configPO.getName(),
+                AlertLevel.getDescriptionByVal(alertLevel));
     }
 
     public StringBuilder getPrefix() {
-        return new StringBuilder(String.format("[Paas Monitor %s告警: %s - %s - %s] [监控项:%s] : [发生时间段: %s ~ %s] ",
+        return new StringBuilder(String.format("[Paas Monitor %s告警: %s - %s - %s] [监控项:%s] : " +
+                        "[发生时间段: %s ~ %s] ",
                 AlertType.getDescriptionByVal(configPO.getType()),
                 SeecSystem.getDescriptionByVal(configPO.getSystemId()), configPO.getCategory(),
                 configPO.getName(), AlertItem.getDescriptionByVal(configPO.getAlertItem()),
@@ -42,7 +44,14 @@ public class AlertMessage {
     }
 
     public String buildUrl() {
-        return "";
+        String type = "";
+        if (configPO.getType() != AlertType.METRIC_COUNT.getVal()
+                && configPO.getType() != AlertType.METRIC_DURATION.getVal()) {
+            type = AlertType.getDescriptionByVal(configPO.getType());
+        } else {
+            type = "metric";
+        }
+        return "https://paas-monitor-ui.seec.seecoder.cn/service/" + type;
     }
 
     public String buildText() {

PaasMonitor-server/src/main/java/com/nju/edu/paasMonitorServer/alert/process/sender/alertmessage/AlertSingle.java

@@ -20,7 +20,7 @@ public class AlertSingle extends AlertMessage {
     public String buildText() {
         StringBuilder text = getPrefix();
         for (SubConditionPO subConditionPO : map.keySet()) {
-            text.append(String.format("[%s阈值:%.2f, 实际值:%.2f] ",
+            text.append(String.format("[%s阈值:%.3f, 实际值:%.3f] ",
                     SubConditionType.getDescriptionByVal(subConditionPO.getType()),
                     subConditionPO.getValue(), map.get(subConditionPO)));
         }

PaasMonitor-server/src/main/java/com/nju/edu/paasMonitorServer/process/MessageConsumer.java

@@ -23,9 +23,7 @@ public class MessageConsumer implements Runnable {
 
     private static volatile MessageConsumer instance;
     private final BlockingQueue<Message> mq;
-    private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE,
-            (int) (CORE_POOL_SIZE * 1.5), 5000,
-            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1000));
+    private final ThreadPoolExecutor threadPoolExecutor;
 
     private int threadPoolFullError = 0;
     private int mqFullError = 0;
@@ -43,6 +41,10 @@ public class MessageConsumer implements Runnable {
 
     private MessageConsumer(BlockingQueue<Message> mq) {
         this.mq = mq;
+        threadPoolExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE,
+                (int) (CORE_POOL_SIZE * 1.5), 5000,
+                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(3000));
+        threadPoolExecutor.allowCoreThreadTimeOut(true);
     }
 
     @Override

sql/paasMonitor.sql

@@ -84,7 +84,7 @@ CREATE TABLE `sub_condition`
     `id`        BIGINT(10) UNSIGNED AUTO_INCREMENT PRIMARY KEY,
     `cid`       BIGINT(10) UNSIGNED NOT NULL,
     `type`      TINYINT(4) UNSIGNED NOT NULL,
-    `value`     DECIMAL(10, 2)      NOT NULL,
+    `value`     DECIMAL(10, 4)      NOT NULL,
     `frequency` INT(4)              NOT NULL DEFAULT 0,
     INDEX `idx_sc_cid` (`cid`)
 ) ENGINE = innoDB