[feat]: migrate blocking executors to virtual threads

hertzbeat-ai/src/main/java/org/apache/hertzbeat/ai/tools/impl/MonitorToolsImpl.java

@@ -21,6 +21,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.ai.config.McpContextHolder;
 import org.apache.hertzbeat.manager.pojo.dto.MonitorDto;
+import org.apache.hertzbeat.manager.pojo.dto.ParamDefineInfo;
 import org.apache.hertzbeat.manager.service.MonitorService;
 import org.apache.hertzbeat.manager.service.AppService;
 import org.apache.hertzbeat.ai.utils.UtilityClass;
@@ -32,7 +33,6 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.apache.hertzbeat.common.entity.manager.Monitor;
 import org.apache.hertzbeat.common.entity.manager.Param;
-import org.apache.hertzbeat.common.entity.manager.ParamDefine;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -280,7 +280,9 @@ public String addMonitor(
 
             // Validate that all required parameters for this monitor type are provided
             try {
-                MonitorDto monitorDto = MonitorDto.builder().monitor(monitor).params(paramList).build();
+                MonitorDto monitorDto = new MonitorDto();
+                monitorDto.setMonitor(monitor);
+                monitorDto.setParams(paramList);
                 monitorService.validate(monitorDto, false);
             } catch (IllegalArgumentException argumentException) {
                 if (argumentException.getMessage().contains("required")) {
@@ -456,7 +458,7 @@ public String getMonitorParams(
             }
 
             // Get parameter definitions from app service
-            List<ParamDefine> paramDefines = appService.getAppParamDefines(app.toLowerCase().trim());
+            List<ParamDefineInfo> paramDefines = appService.getAppParamDefines(app.toLowerCase().trim());
 
             if (paramDefines == null || paramDefines.isEmpty()) {
                 return String.format("No parameter definitions found for monitor type '%s'. "
@@ -468,7 +470,7 @@ public String getMonitorParams(
             response.append(String.format("Parameter Definitions for Monitor Type '%s' (Total: %d):\n\n",
                 app, paramDefines.size()));
 
-            for (ParamDefine paramDefine : paramDefines) {
+            for (ParamDefineInfo paramDefine : paramDefines) {
                 response.append("• Field: ").append(paramDefine.getField()).append("\n");
 
                 // Add display name if available

hertzbeat-alerter/pom.xml

@@ -40,6 +40,11 @@
           <artifactId>hertzbeat-common-core</artifactId>
           <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hertzbeat</groupId>
+            <artifactId>hertzbeat-common-spring</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <!-- plugin -->
         <dependency>
             <groupId>org.apache.hertzbeat</groupId>

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

@@ -18,29 +18,46 @@
 package org.apache.hertzbeat.alert;
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.concurrent.ManagedExecutor;
+import org.apache.hertzbeat.common.concurrent.ManagedExecutors;
+import org.apache.hertzbeat.common.config.VirtualThreadProperties;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 /**
  * alarm module thread pool
  */
 @Component
 @Slf4j
-public class AlerterWorkerPool {
+public class AlerterWorkerPool implements DisposableBean {
 
     private ThreadPoolExecutor workerExecutor;
-    private ThreadPoolExecutor notifyExecutor;
-    private ThreadPoolExecutor logWorkerExecutor;
+    private ManagedExecutor notifyExecutor;
+    private ManagedExecutor logWorkerExecutor;
+    private Map<Byte, Semaphore> notifyChannelPermits;
+    private int notifyMaxConcurrentPerChannel;
 
     public AlerterWorkerPool() {
+        this(VirtualThreadProperties.defaults());
+    }
+
+    @Autowired
+    public AlerterWorkerPool(VirtualThreadProperties virtualThreadProperties) {
+        VirtualThreadProperties properties =
+                virtualThreadProperties == null ? VirtualThreadProperties.defaults() : virtualThreadProperties;
         initWorkExecutor();
-        initNotifyExecutor();
-        initLogWorkerExecutor();
+        initNotifyExecutor(properties);
+        initLogWorkerExecutor(properties);
     }
 
     private void initWorkExecutor() {
@@ -61,16 +78,32 @@ private void initWorkExecutor() {
                 new ThreadPoolExecutor.AbortPolicy());
     }
 
-    private void initNotifyExecutor() {
+    private void initNotifyExecutor(VirtualThreadProperties properties) {
+        Thread.UncaughtExceptionHandler handler = (thread, throwable) -> {
+            log.error("Alerter notifyExecutor has uncaughtException.");
+            log.error(throwable.getMessage(), throwable);
+        };
+        if (properties.enabled()) {
+            VirtualThreadProperties.AlerterProperties alerterProperties = properties.alerter();
+            VirtualThreadProperties.PoolProperties notifyProperties = alerterProperties.notifyPool();
+            notifyMaxConcurrentPerChannel = Math.max(1, alerterProperties.notifyMaxConcurrentPerChannel());
+            notifyChannelPermits = new ConcurrentHashMap<>(8);
+            notifyExecutor = ManagedExecutors.newVirtualExecutor("notify-worker", "notify-worker-",
+                    notifyProperties.mode(), notifyProperties.maxConcurrentJobs(), handler);
+            return;
+        }
+        notifyMaxConcurrentPerChannel = 0;
+        notifyChannelPermits = null;
+        notifyExecutor = ManagedExecutors.wrap("notify-worker", createLegacyNotifyExecutor(handler));
+    }
+
+    private ThreadPoolExecutor createLegacyNotifyExecutor(Thread.UncaughtExceptionHandler handler) {
         ThreadFactory threadFactory = new ThreadFactoryBuilder()
-                .setUncaughtExceptionHandler((thread, throwable) -> {
-                    log.error("Alerter notifyExecutor has uncaughtException.");
-                    log.error(throwable.getMessage(), throwable);
-                })
+                .setUncaughtExceptionHandler(handler)
                 .setDaemon(true)
                 .setNameFormat("notify-worker-%d")
                 .build();
-        notifyExecutor = new ThreadPoolExecutor(6,
+        return new ThreadPoolExecutor(6,
                 6,
                 10,
                 TimeUnit.SECONDS,
@@ -79,16 +112,27 @@ private void initNotifyExecutor() {
                 new ThreadPoolExecutor.AbortPolicy());
     }
 
-    private void initLogWorkerExecutor() {
+    private void initLogWorkerExecutor(VirtualThreadProperties properties) {
+        Thread.UncaughtExceptionHandler handler = (thread, throwable) -> {
+            log.error("Alerter logWorkerExecutor has uncaughtException.");
+            log.error(throwable.getMessage(), throwable);
+        };
+        if (properties.enabled()) {
+            VirtualThreadProperties.QueueProperties logWorkerProperties = properties.alerter().logWorker();
+            logWorkerExecutor = ManagedExecutors.newQueuedVirtualExecutor("alerter-log-worker", "log-worker-",
+                    logWorkerProperties.maxConcurrentJobs(), logWorkerProperties.queueCapacity(), handler);
+            return;
+        }
+        logWorkerExecutor = ManagedExecutors.wrap("alerter-log-worker", createLegacyLogWorkerExecutor(handler));
+    }
+
+    private ThreadPoolExecutor createLegacyLogWorkerExecutor(Thread.UncaughtExceptionHandler handler) {
         ThreadFactory threadFactory = new ThreadFactoryBuilder()
-                .setUncaughtExceptionHandler((thread, throwable) -> {
-                    log.error("Alerter logWorkerExecutor has uncaughtException.");
-                    log.error(throwable.getMessage(), throwable);
-                })
+                .setUncaughtExceptionHandler(handler)
                 .setDaemon(true)
                 .setNameFormat("log-worker-%d")
                 .build();
-        logWorkerExecutor = new ThreadPoolExecutor(10, 10, 10, TimeUnit.SECONDS,
+        return new ThreadPoolExecutor(10, 10, 10, TimeUnit.SECONDS,
                 new LinkedBlockingQueue<>(1000),
                 threadFactory,
                 new ThreadPoolExecutor.AbortPolicy());
@@ -113,6 +157,41 @@ public void executeNotify(Runnable runnable) throws RejectedExecutionException {
         notifyExecutor.execute(runnable);
     }
 
+    /**
+     * Executes the given runnable task using the notify executor with per-channel concurrency control.
+     *
+     * @param channelType notification channel type
+     * @param runnable the task to be executed
+     * @throws RejectedExecutionException if the task cannot be accepted for execution
+     */
+    public void executeNotify(byte channelType, Runnable runnable) throws RejectedExecutionException {
+        if (notifyChannelPermits == null) {
+            notifyExecutor.execute(runnable);
+            return;
+        }
+        Semaphore semaphore = notifyChannelPermits.computeIfAbsent(channelType,
+                key -> new Semaphore(notifyMaxConcurrentPerChannel));
+        if (!semaphore.tryAcquire()) {
+            throw new RejectedExecutionException(
+                    "notify-worker rejected task because channel concurrency limit was reached for type " + channelType);
+        }
+        boolean submitted = false;
+        try {
+            notifyExecutor.execute(() -> {
+                try {
+                    runnable.run();
+                } finally {
+                    semaphore.release();
+                }
+            });
+            submitted = true;
+        } finally {
+            if (!submitted) {
+                semaphore.release();
+            }
+        }
+    }
+
     /**
      * Executes the given runnable task using the logWorkerExecutor.
      *
@@ -122,4 +201,11 @@ public void executeNotify(Runnable runnable) throws RejectedExecutionException {
     public void executeLogJob(Runnable runnable) throws RejectedExecutionException {
         logWorkerExecutor.execute(runnable);
     }
+
+    @Override
+    public void destroy() {
+        workerExecutor.shutdownNow();
+        notifyExecutor.close();
+        logWorkerExecutor.close();
+    }
 }