add AgentAware

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

@@ -51,6 +51,11 @@ public class ThreadPoolStats extends Metrics {
      */
     private int maximumPoolSize;
 
+    /**
+     * 空闲时间 (ms)
+     */
+    private long keepAliveTime;
+
     /**
      * 队列类型
      */

core/src/main/java/org/dromara/dynamictp/core/aware/AwareManager.java

@@ -96,7 +96,7 @@ public static void execute(Executor executor, Runnable r) {
     public static void beforeExecute(Executor executor, Thread t, Runnable r) {
         for (ExecutorAware aware : EXECUTOR_AWARE_LIST) {
             try {
-                aware.beforeExecute(executor, t, r);
+                r = aware.beforeExecuteWrap(executor, t, r);
             } catch (Exception e) {
                 log.error("DynamicTp aware [{}], enhance beforeExecute error.", aware.getName(), e);
             }
@@ -106,7 +106,7 @@ public static void beforeExecute(Executor executor, Thread t, Runnable r) {
     public static void afterExecute(Executor executor, Runnable r, Throwable t) {
         for (ExecutorAware aware : EXECUTOR_AWARE_LIST) {
             try {
-                aware.afterExecute(executor, r, t);
+                r = aware.afterExecuteWrap(executor, r, t);
             } catch (Exception e) {
                 log.error("DynamicTp aware [{}], enhance afterExecute error.", aware.getName(), e);
             }
@@ -146,7 +146,7 @@ public static void terminated(Executor executor) {
     public static void beforeReject(Runnable r, Executor executor) {
         for (ExecutorAware aware : EXECUTOR_AWARE_LIST) {
             try {
-                aware.beforeReject(r, executor);
+                r = aware.beforeRejectWrap(r, executor);
             } catch (Exception e) {
                 log.error("DynamicTp aware [{}], enhance beforeReject error.", aware.getName(), e);
             }
@@ -156,7 +156,7 @@ public static void beforeReject(Runnable r, Executor executor) {
     public static void afterReject(Runnable r, Executor executor) {
         for (ExecutorAware aware : EXECUTOR_AWARE_LIST) {
             try {
-                aware.afterReject(r, executor);
+                r = aware.afterRejectWrap(r, executor);
             } catch (Exception e) {
                 log.error("DynamicTp aware [{}], enhance afterReject error.", aware.getName(), e);
             }

core/src/main/java/org/dromara/dynamictp/core/aware/ExecutorAware.java

@@ -91,6 +91,11 @@ default void beforeExecute(Executor executor, Thread t, Runnable r) {
         // default no Operation
     }
 
+    default Runnable beforeExecuteWrap(Executor executor, Thread t, Runnable r) {
+        beforeExecute(executor, t, r);
+        return r;
+    }
+
     /**
      * enhance afterExecute
      *
@@ -102,6 +107,11 @@ default void afterExecute(Executor executor, Runnable r, Throwable t) {
         // default no Operation
     }
 
+    default Runnable afterExecuteWrap(Executor executor, Runnable r, Throwable t) {
+        afterExecute(executor, r, t);
+        return r;
+    }
+
     /**
      * enhance shutdown
      *
@@ -139,6 +149,11 @@ default void beforeReject(Runnable r, Executor executor) {
         // default no Operation
     }
 
+    default Runnable beforeRejectWrap(Runnable r, Executor executor) {
+        beforeReject(r, executor);
+        return r;
+    }
+
     /**
      * enhance after reject
      * @param r runnable
@@ -147,4 +162,9 @@ default void beforeReject(Runnable r, Executor executor) {
     default void afterReject(Runnable r, Executor executor) {
         // default no Operation
     }
+
+    default Runnable afterRejectWrap(Runnable r, Executor executor) {
+        afterReject(r, executor);
+        return r;
+    }
 }

core/src/main/java/org/dromara/dynamictp/core/converter/ExecutorConverter.java

@@ -97,6 +97,7 @@ private static ThreadPoolStats convertCommon(ExecutorAdapter<?> executor) {
         poolStats.setCompletedTaskCount(executor.getCompletedTaskCount());
         poolStats.setWaitTaskCount(executor.getQueueSize());
         poolStats.setRejectHandlerName(executor.getRejectHandlerType());
+        poolStats.setKeepAliveTime(executor.getKeepAliveTime(TimeUnit.MILLISECONDS));
         return poolStats;
     }
 }

core/src/main/java/org/dromara/dynamictp/core/monitor/collector/jmx/JMXCollector.java

@@ -21,14 +21,11 @@
 import org.dromara.dynamictp.common.em.CollectorTypeEnum;
 import org.dromara.dynamictp.common.entity.ThreadPoolStats;
 import org.dromara.dynamictp.core.monitor.collector.AbstractCollector;
-import org.springframework.beans.BeanUtils;
 
 import javax.management.JMException;
 import javax.management.MBeanServer;
 import javax.management.ObjectName;
 import java.lang.management.ManagementFactory;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * ThreadPoolStatsInfo related
@@ -40,26 +37,15 @@ public class JMXCollector extends AbstractCollector {
 
     public static final String DTP_METRIC_NAME_PREFIX = "dtp.thread.pool";
 
-    /**
-     * thread pool stats map
-     */
-    private static final Map<String, ThreadPoolStats> GAUGE_CACHE = new ConcurrentHashMap<>();
-
     @Override
     public void collect(ThreadPoolStats threadPoolStats) {
-        if (GAUGE_CACHE.containsKey(threadPoolStats.getPoolName())) {
-            ThreadPoolStats poolStats = GAUGE_CACHE.get(threadPoolStats.getPoolName());
-            BeanUtils.copyProperties(threadPoolStats, poolStats);
-        } else {
-            try {
-                MBeanServer server = ManagementFactory.getPlatformMBeanServer();
-                ObjectName name = new ObjectName(DTP_METRIC_NAME_PREFIX + ":name=" + threadPoolStats.getPoolName());
-                ThreadPoolStatsJMX stats = new ThreadPoolStatsJMX(threadPoolStats);
-                server.registerMBean(stats, name);
-            } catch (JMException e) {
-                log.error("collect thread pool stats error", e);
-            }
-            GAUGE_CACHE.put(threadPoolStats.getPoolName(), threadPoolStats);
+        try {
+            MBeanServer server = ManagementFactory.getPlatformMBeanServer();
+            ObjectName name = new ObjectName(DTP_METRIC_NAME_PREFIX + ":name=" + threadPoolStats.getPoolName());
+            ThreadPoolStatsJMX stats = new ThreadPoolStatsJMX(threadPoolStats);
+            server.registerMBean(stats, name);
+        } catch (JMException e) {
+            log.error("collect thread pool stats error", e);
         }
     }
 

dependencies/pom.xml

@@ -12,7 +12,7 @@
     <url>https://github.com/yanhom1314/dynamic-tp</url>
 
     <properties>
-        <revision>1.1.7</revision>
+        <revision>1.1.8-beta</revision>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 
         <lombok.version>1.18.24</lombok.version>
@@ -426,6 +426,12 @@
                 <version>${revision}</version>
             </dependency>
 
+            <dependency>
+                <groupId>org.dromara.dynamictp</groupId>
+                <artifactId>dynamic-tp-extension-agent</artifactId>
+                <version>${revision}</version>
+            </dependency>
+
             <dependency>
                 <groupId>org.dromara.dynamictp</groupId>
                 <artifactId>dynamic-tp-extension-notify-email</artifactId>

extension/extension-agent/pom.xml

@@ -0,0 +1,14 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.dromara.dynamictp</groupId>
+        <artifactId>dynamic-tp-extension</artifactId>
+        <version>${revision}</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>dynamic-tp-extension-agent</artifactId>
+
+    <dependencies>
+    </dependencies>
+</project>

extension/extension-agent/src/main/java/org/dromara/dynamictp/core/aware/AgentAware.java

@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.dromara.dynamictp.core.aware;
+
+import cn.hutool.core.util.ArrayUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.dromara.dynamictp.core.support.task.runnable.DtpRunnable;
+
+import java.lang.ref.SoftReference;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+
+import static org.dromara.dynamictp.common.constant.DynamicTpConst.DTP_EXECUTE_ENHANCED;
+import static org.dromara.dynamictp.common.constant.DynamicTpConst.FALSE_STR;
+
+/**
+ * deal agent wrapper
+ * @author txbao
+ */
+@Slf4j
+@SuppressWarnings("all")
+public class AgentAware extends TaskStatAware {
+
+    /**
+     * dtpRunnableCache  key -> Runnable  value -> DtpRunnable
+     */
+    private final Map<Runnable, SoftReference<DtpRunnable>> dtpRunnableCache = new ConcurrentHashMap<>();
+
+    @Override
+    public int getOrder() {
+        return Integer.MIN_VALUE;
+    }
+
+    @Override
+    public String getName() {
+        return "agent";
+    }
+
+    private DtpRunnable determineDtpRunnable(List<Field> conditionalFields, Runnable r) throws IllegalAccessException {
+        for (Field field : conditionalFields) {
+            if (Objects.isNull(field)) {
+                continue;
+            }
+            field.setAccessible(true);
+            Runnable o = (Runnable) field.get(r);
+            if (o instanceof DtpRunnable) {
+                return (DtpRunnable) o;
+            }
+            // 纵向查找
+            DtpRunnable dtpRunnable = getDtpRunnable(o.getClass(), o);
+            if (dtpRunnable != null) {
+                return dtpRunnable;
+            }
+        }
+        return null;
+    }
+
+    private DtpRunnable getDtpRunnable(Class<? extends Runnable> rClass, Runnable r) throws IllegalAccessException {
+        while (Runnable.class.isAssignableFrom(rClass)) {
+            Field[] declaredFields = rClass.getDeclaredFields();
+            if (ArrayUtil.isNotEmpty(declaredFields)) {
+                List<Field> conditionFields = Arrays.stream(declaredFields)
+                        .filter(ele -> Runnable.class.isAssignableFrom(ele.getType()))
+                        .collect(Collectors.toList());
+                if (CollectionUtils.isNotEmpty(conditionFields)) {
+                    DtpRunnable dtpRunnable = determineDtpRunnable(conditionFields, r);
+                    if (Objects.nonNull(dtpRunnable)) {
+                        return dtpRunnable;
+                    }
+                }
+            }
+            if (!Runnable.class.isAssignableFrom(rClass.getSuperclass())) {
+                break;
+            }
+            rClass = (Class<? extends Runnable>) rClass.getSuperclass();
+        }
+        return null;
+    }
+
+    private Runnable getDtpRunnableInstance(Runnable r) {
+        if (r instanceof DtpRunnable) {
+            return r;
+        }
+        DtpRunnable dtpRunnable = null;
+        Class<? extends Runnable> rClass = r.getClass();
+        try {
+            dtpRunnable = getDtpRunnable(rClass, r);
+        } catch (IllegalAccessException e) {
+            log.error("getDtpRunnable Error", e);
+        }
+        if (dtpRunnable == null) {
+            if (log.isWarnEnabled()) {
+                log.warn("DynamicTp aware [{}], can not find DtpRunnable.", getName());
+            }
+            return r;
+        }
+        return dtpRunnable;
+    }
+
+    @Override
+    public Runnable beforeExecuteWrap(Executor executor, Thread t, Runnable r) {
+        Runnable runnableWrap = getDtpRunnableInstance(r);
+        if (runnableWrap instanceof DtpRunnable) {
+            dtpRunnableCache.put(r, new SoftReference<>((DtpRunnable) runnableWrap));
+        } else {
+            // 被封装的wrapper没有找到DtpRunnable对象，那么就关闭某些监控指标，防止内存溢出
+            System.setProperty(DTP_EXECUTE_ENHANCED, FALSE_STR);
+        }
+        return runnableWrap;
+    }
+
+    @Override
+    public Runnable afterExecuteWrap(Executor executor, Runnable r, Throwable t) {
+        SoftReference<DtpRunnable> remove = dtpRunnableCache.remove(r);
+        if (remove != null) {
+            return remove.get();
+        }
+        return getDtpRunnableInstance(r);
+    }
+
+    @Override
+    public Runnable beforeRejectWrap(Runnable r, Executor executor) {
+        SoftReference<DtpRunnable> remove = dtpRunnableCache.remove(r);
+        if (remove != null) {
+            return remove.get();
+        }
+        return getDtpRunnableInstance(r);
+    }
+}

extension/extension-agent/src/main/resources/META-INF/services/org.dromara.dynamictp.core.aware.ExecutorAware

@@ -0,0 +1 @@
+org.dromara.dynamictp.core.aware.AgentAware

extension/extension-skywalking/pom.xml

@@ -21,11 +21,6 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
-
-        <dependency>
-            <groupId>org.dromara.dynamictp</groupId>
-            <artifactId>dynamic-tp-jvmti-runtime</artifactId>
-        </dependency>
     </dependencies>
 
 </project>