[ISSUE #442] 重构告警规则

common/src/main/java/org/dromara/dynamictp/common/entity/NotifyItem.java

@@ -58,6 +58,11 @@ public class NotifyItem {
      */
     private int threshold;
 
+    /**
+     * The count to alarm.
+     */
+    private int count = 1;
+
     /**
      * Alarm interval, time unit（s）
      */

core/src/main/java/org/dromara/dynamictp/core/notifier/chain/filter/AlarmBaseFilter.java

@@ -56,20 +56,25 @@ public void doFilter(BaseNotifyCtx context, Invoker<BaseNotifyCtx> nextInvoker)
             return;
         }
 
-        if (!AlarmManager.checkThreshold(executorWrapper, context.getNotifyItemEnum(), notifyItem)) {
-            return;
-        }
-        synchronized (SEND_LOCK) {
-            // recheck alarm limit.
-            ifAlarm = AlarmLimiter.ifAlarm(executorWrapper.getThreadPoolName(), notifyItem.getType());
-            if (!ifAlarm) {
-                log.warn("DynamicTp notify, concurrent send, alarm limit, threadPoolName: {}, notifyItem: {}",
-                        executorWrapper.getThreadPoolName(), notifyItem.getType());
+        if (AlarmManager.checkThreshold(executorWrapper, context.getNotifyItemEnum(), notifyItem)) {
+
+            if (!AlarmManager.incrementAndCheckAlarmCount(executorWrapper, context.getNotifyItemEnum(), notifyItem)) {
                 return;
             }
-            AlarmLimiter.putVal(executorWrapper.getThreadPoolName(), notifyItem.getType());
+
+            synchronized (SEND_LOCK) {
+                // recheck alarm limit.
+                ifAlarm = AlarmLimiter.ifAlarm(executorWrapper.getThreadPoolName(), notifyItem.getType());
+                if (!ifAlarm) {
+                    log.warn("DynamicTp notify, concurrent send, alarm limit, threadPoolName: {}, notifyItem: {}",
+                            executorWrapper.getThreadPoolName(), notifyItem.getType());
+                    return;
+                }
+                AlarmManager.resetAlarmCount(executorWrapper, context.getNotifyItemEnum());
+                AlarmLimiter.putVal(executorWrapper.getThreadPoolName(), notifyItem.getType());
+            }
+            nextInvoker.invoke(context);
         }
-        nextInvoker.invoke(context);
     }
 
     private boolean satisfyBaseCondition(NotifyItem notifyItem, ExecutorWrapper executor) {

core/src/main/java/org/dromara/dynamictp/core/notifier/manager/AlarmManager.java

@@ -37,7 +37,10 @@
 import org.slf4j.MDC;
 
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.dromara.dynamictp.common.constant.DynamicTpConst.TRACE_ID;
 import static org.dromara.dynamictp.common.em.QueueTypeEnum.LINKED_BLOCKING_QUEUE;
@@ -67,6 +70,8 @@ public class AlarmManager {
         ALARM_INVOKER_CHAIN = NotifyFilterBuilder.getAlarmInvokerChain();
     }
 
+    private static final Map<String, AtomicInteger> ALARM_COUNTER = new ConcurrentHashMap<>();
+
     private AlarmManager() { }
 
     public static void initAlarm(String poolName, List<NotifyItem> notifyItems) {
@@ -116,6 +121,16 @@ public static boolean checkThreshold(ExecutorWrapper executor, NotifyItemEnum no
         }
     }
 
+    public static boolean incrementAndCheckAlarmCount(ExecutorWrapper executorWrapper, NotifyItemEnum notifyType, NotifyItem notifyItem) {
+        String key = genKey(executorWrapper.getThreadPoolName(), notifyType.getValue());
+        return ALARM_COUNTER.computeIfAbsent(key, k -> new AtomicInteger(0)).incrementAndGet() >= notifyItem.getCount();
+    }
+
+    public static void resetAlarmCount(ExecutorWrapper executorWrapper, NotifyItemEnum notifyType) {
+        String key = genKey(executorWrapper.getThreadPoolName(), notifyType.getValue());
+        ALARM_COUNTER.getOrDefault(key, new AtomicInteger(0)).set(0);
+    }
+
     public static void destroy() {
         ALARM_EXECUTOR.shutdownNow();
     }
@@ -153,4 +168,8 @@ private static void postAlarm(Runnable runnable) {
             MDC.remove(TRACE_ID);
         }
     }
+
+    private static String genKey(String threadPoolName, String type) {
+        return threadPoolName + ":" + type;
+    }
 }