[Improve] Refactor the notification sending logic.

alerter/src/main/java/org/apache/hertzbeat/alert/AlerterWorkerPool.java

@@ -34,9 +34,11 @@
 public class AlerterWorkerPool {
 
     private ThreadPoolExecutor workerExecutor;
+    private ThreadPoolExecutor notifyExecutor;
 
     public AlerterWorkerPool() {
         initWorkExecutor();
+        initNotifyExecutor();
     }
 
     private void initWorkExecutor() {
@@ -57,6 +59,24 @@ private void initWorkExecutor() {
                 new ThreadPoolExecutor.AbortPolicy());
     }
 
+    private void initNotifyExecutor() {
+        ThreadFactory threadFactory = new ThreadFactoryBuilder()
+                .setUncaughtExceptionHandler((thread, throwable) -> {
+                    log.error("notifyExecutor has uncaughtException.");
+                    log.error(throwable.getMessage(), throwable);
+                })
+                .setDaemon(true)
+                .setNameFormat("notify-worker-%d")
+                .build();
+        notifyExecutor = new ThreadPoolExecutor(6,
+                10,
+                10,
+                TimeUnit.SECONDS,
+                new SynchronousQueue<>(),
+                threadFactory,
+                new ThreadPoolExecutor.AbortPolicy());
+    }
+
     /**
      * Run the alerter task
      * @param runnable task
@@ -65,4 +85,15 @@ private void initWorkExecutor() {
     public void executeJob(Runnable runnable) throws RejectedExecutionException {
         workerExecutor.execute(runnable);
     }
+
+    /**
+     * Executes the given runnable task using the notifyExecutor.
+     *
+     * @param runnable the task to be executed
+     * @throws RejectedExecutionException if the task cannot be accepted for execution
+     */
+    public void executeNotify(Runnable runnable) throws RejectedExecutionException {
+        notifyExecutor.execute(runnable);
+    }
+
 }

manager/src/main/java/org/apache/hertzbeat/manager/component/alerter/DispatcherAlarm.java

@@ -20,6 +20,7 @@
 import com.google.common.collect.Maps;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.ServiceLoader;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.alert.AlerterWorkerPool;
@@ -99,8 +100,8 @@ private NoticeTemplate getOneTemplateById(Long id) {
         return noticeConfigService.getOneTemplateById(id);
     }
 
-    private List<NoticeRule> matchNoticeRulesByAlert(Alert alert) {
-        return noticeConfigService.getReceiverFilterRule(alert);
+    private Optional<List<NoticeRule>> matchNoticeRulesByAlert(Alert alert) {
+        return Optional.ofNullable(noticeConfigService.getReceiverFilterRule(alert));
     }
 
     private class DispatchTask implements Runnable {
@@ -131,29 +132,21 @@ public void run() {
         }
 
         private void sendNotify(Alert alert) {
-            List<NoticeRule> noticeRules = matchNoticeRulesByAlert(alert);
-            // todo Send notification here temporarily single thread
-            if (noticeRules != null) {
-                for (NoticeRule rule : noticeRules) {
-                    try {
-                        if (rule.getTemplateId() == null) {
-                            List<Long> receiverIdList = rule.getReceiverId();
-                            for (Long receiverId : receiverIdList) {
-                                sendNoticeMsg(getOneReceiverById(receiverId),
-                                        null, alert);
-                            }
-                        } else {
-                            List<Long> receiverIdList = rule.getReceiverId();
-                            for (Long receiverId : receiverIdList) {
-                                sendNoticeMsg(getOneReceiverById(receiverId),
-                                    getOneTemplateById(rule.getTemplateId()), alert);
-                            }
-                        }
-                    } catch (AlertNoticeException e) {
-                        log.warn("DispatchTask sendNoticeMsg error, message: {}", e.getMessage());
-                    }
-                }
-            }
+            matchNoticeRulesByAlert(alert).ifPresent(noticeRules -> {
+                noticeRules.forEach(rule -> {
+                    workerPool.executeNotify(() -> {
+                        rule.getReceiverId().stream().filter(receiverId -> rule.getTemplateId() != null)
+                                .forEach(receiverId -> {
+                                    try {
+                                        sendNoticeMsg(getOneReceiverById(receiverId),
+                                                getOneTemplateById(rule.getTemplateId()), alert);
+                                    } catch (AlertNoticeException e) {
+                                        log.warn("DispatchTask sendNoticeMsg error, message: {}", e.getMessage());
+                                    }
+                                });
+                    });
+                });
+            });
         }
     }