feat: 线程池变更配置添加拒绝策略.

common/src/main/java/io/dynamic/threadpool/common/model/PoolParameter.java

@@ -64,6 +64,13 @@ public interface PoolParameter {
      */
     Integer getKeepAliveTime();
 
+    /**
+     * rejectedType
+     *
+     * @return
+     */
+    Integer getRejectedType();
+
     /**
      * isAlarm
      *

common/src/main/java/io/dynamic/threadpool/common/model/PoolParameterInfo.java

@@ -60,6 +60,11 @@ public class PoolParameterInfo implements PoolParameter, Serializable {
      */
     private Integer keepAliveTime;
 
+    /**
+     * 拒绝策略类型
+     */
+    private Integer rejectedType;
+
     /**
      * 是否告警
      */

common/src/main/java/io/dynamic/threadpool/common/toolkit/ContentUtil.java

@@ -40,6 +40,7 @@ public class ContentUtil {
         poolInfo.setIsAlarm(parameter.getIsAlarm());
         poolInfo.setCapacityAlarm(parameter.getCapacityAlarm());
         poolInfo.setLivenessAlarm(parameter.getLivenessAlarm());
+        poolInfo.setRejectedType(parameter.getRejectedType());
         return JSON.toJSONString(poolInfo);
     }
 

config/src/main/java/io/dynamic/threadpool/config/model/ConfigInfoBase.java

@@ -64,6 +64,11 @@ public class ConfigInfoBase implements Serializable {
      */
     private Integer keepAliveTime;
 
+    /**
+     * 拒绝策略类型
+     */
+    private Integer rejectedType;
+
     /**
      * 是否告警
      */

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/core/ThreadPoolDynamicRefresh.java

@@ -2,9 +2,12 @@ package io.dynamic.threadpool.starter.core;
 
 import com.alibaba.fastjson.JSON;
 import io.dynamic.threadpool.common.model.PoolParameterInfo;
-import io.dynamic.threadpool.starter.handler.ThreadPoolChangeHandler;
+import io.dynamic.threadpool.starter.toolkit.thread.QueueTypeEnum;
+import io.dynamic.threadpool.starter.toolkit.thread.RejectedTypeEnum;
+import io.dynamic.threadpool.starter.toolkit.thread.ResizableCapacityLinkedBlockIngQueue;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Objects;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
@@ -22,23 +25,60 @@ public class ThreadPoolDynamicRefresh {
         String tpId = parameter.getTpId();
         Integer coreSize = parameter.getCoreSize(), maxSize = parameter.getMaxSize(),
                 queueType = parameter.getQueueType(), capacity = parameter.getCapacity(),
-                keepAliveTime = parameter.getKeepAliveTime();
-        refreshDynamicPool(tpId, coreSize, maxSize, queueType, capacity, keepAliveTime);
+                keepAliveTime = parameter.getKeepAliveTime(), rejectedType = parameter.getRejectedType();
+        refreshDynamicPool(tpId, coreSize, maxSize, queueType, capacity, keepAliveTime, rejectedType);
     }
 
-    public static void refreshDynamicPool(String threadPoolId, Integer coreSize, Integer maxSize, Integer queueType, Integer capacity, Integer keepAliveTime) {
+    public static void refreshDynamicPool(String threadPoolId, Integer coreSize, Integer maxSize, Integer queueType, Integer capacity, Integer keepAliveTime, Integer rejectedType) {
         ThreadPoolExecutor executor = GlobalThreadPoolManage.getExecutorService(threadPoolId).getPool();
         printLog("[🔥] Original thread pool. ",
-                executor.getCorePoolSize(), executor.getMaximumPoolSize(), queueType, executor.getQueue().remainingCapacity(), executor.getKeepAliveTime(TimeUnit.MILLISECONDS));
+                executor.getCorePoolSize(),
+                executor.getMaximumPoolSize(),
+                queueType,
+                (executor.getQueue().remainingCapacity() + executor.getQueue().size()),
+                executor.getKeepAliveTime(TimeUnit.MILLISECONDS),
+                rejectedType);
 
-        ThreadPoolChangeHandler.changePool(executor, coreSize, maxSize, queueType, capacity, keepAliveTime);
+        changePoolInfo(executor, coreSize, maxSize, queueType, capacity, keepAliveTime, rejectedType);
         ThreadPoolExecutor afterExecutor = GlobalThreadPoolManage.getExecutorService(threadPoolId).getPool();
 
         printLog("[🚀] Changed thread pool. ",
-                afterExecutor.getCorePoolSize(), afterExecutor.getMaximumPoolSize(), queueType, afterExecutor.getQueue().remainingCapacity(), afterExecutor.getKeepAliveTime(TimeUnit.MILLISECONDS));
+                afterExecutor.getCorePoolSize(),
+                afterExecutor.getMaximumPoolSize(),
+                queueType,
+                (afterExecutor.getQueue().remainingCapacity() + afterExecutor.getQueue().size()),
+                afterExecutor.getKeepAliveTime(TimeUnit.MILLISECONDS),
+                rejectedType);
     }
 
-    private static void printLog(String prefixMsg, Integer coreSize, Integer maxSize, Integer queueType, Integer capacity, Long keepAliveTime) {
-        log.info("{} coreSize :: {}, maxSize :: {}, queueType :: {}, capacity :: {}, keepAliveTime :: {}", prefixMsg, coreSize, maxSize, queueType, capacity, keepAliveTime);
+    private static void printLog(String prefixMsg, Integer coreSize, Integer maxSize, Integer queueType, Integer capacity, Long keepAliveTime, Integer rejectedType) {
+        log.info("{} coreSize :: {}, maxSize :: {}, queueType :: {}, capacity :: {}, keepAliveTime :: {}, rejectedType:: {}", prefixMsg, coreSize, maxSize, queueType, capacity, keepAliveTime, rejectedType);
+    }
+
+    public static void changePoolInfo(ThreadPoolExecutor executor, Integer coreSize, Integer maxSize, Integer queueType, Integer capacity, Integer keepAliveTime, Integer rejectedType) {
+        if (coreSize != null) {
+            executor.setCorePoolSize(coreSize);
+        }
+
+        if (maxSize != null) {
+            executor.setMaximumPoolSize(maxSize);
+        }
+
+        if (capacity != null && Objects.equals(QueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.type, queueType)) {
+            if (executor.getQueue() instanceof ResizableCapacityLinkedBlockIngQueue) {
+                ResizableCapacityLinkedBlockIngQueue queue = (ResizableCapacityLinkedBlockIngQueue) executor.getQueue();
+                queue.setCapacity(capacity);
+            } else {
+                log.warn("[Pool change] The queue length cannot be modified. Queue type mismatch. Current queue type :: {}", executor.getQueue().getClass().getSimpleName());
+            }
+        }
+
+        if (keepAliveTime != null) {
+            executor.setKeepAliveTime(keepAliveTime, TimeUnit.SECONDS);
+        }
+
+        if (rejectedType != null) {
+            executor.setRejectedExecutionHandler(RejectedTypeEnum.createPolicy(queueType));
+        }
     }
 }

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/core/ThreadPoolRunStateHandler.java

@@ -1,7 +1,6 @@
-package io.dynamic.threadpool.starter.handler;
+package io.dynamic.threadpool.starter.core;
 
 import io.dynamic.threadpool.common.model.PoolRunStateInfo;
-import io.dynamic.threadpool.starter.core.GlobalThreadPoolManage;
 import io.dynamic.threadpool.starter.wrap.CustomThreadPoolExecutor;
 import io.dynamic.threadpool.starter.wrap.DynamicThreadPoolWrap;
 
@@ -11,10 +10,10 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 
 /**
- * 线程池运行状态组件.
+ * Thread Pool Run State Service.
  *
  * @author chen.ma
- * @date 2021/7/7 19:37
+ * @date 2021/7/12 21:25
  */
 public class ThreadPoolRunStateHandler {
 
@@ -87,5 +86,4 @@ public class ThreadPoolRunStateHandler {
     private static String divide(int num1, int num2) {
         return ((int) (Double.parseDouble(num1 + "") / Double.parseDouble(num2 + "") * 100)) + "%";
     }
-
 }

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/handler/ThreadPoolChangeHandler.java

@@ -1,42 +0,0 @@
-package io.dynamic.threadpool.starter.handler;
-
-import io.dynamic.threadpool.starter.toolkit.thread.QueueTypeEnum;
-import io.dynamic.threadpool.starter.toolkit.thread.ResizableCapacityLinkedBlockIngQueue;
-import lombok.extern.slf4j.Slf4j;
-
-import java.util.Objects;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-/**
- * 线程池修改工具类
- *
- * @author chen.ma
- * @date 2021/6/25 17:19
- */
-@Slf4j
-public class ThreadPoolChangeHandler {
-
-    public static void changePool(ThreadPoolExecutor executor, Integer coreSize, Integer maxSize, Integer queueType, Integer capacity, Integer keepAliveTime) {
-        if (coreSize != null) {
-            executor.setCorePoolSize(coreSize);
-        }
-
-        if (maxSize != null) {
-            executor.setMaximumPoolSize(maxSize);
-        }
-
-        if (capacity != null && Objects.equals(QueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.type, queueType)) {
-            if (executor.getQueue() instanceof ResizableCapacityLinkedBlockIngQueue) {
-                ResizableCapacityLinkedBlockIngQueue queue = (ResizableCapacityLinkedBlockIngQueue) executor.getQueue();
-                queue.setCapacity(capacity);
-            } else {
-                log.warn("[Pool change] The queue length cannot be modified. Queue type mismatch.");
-            }
-        }
-
-        if (keepAliveTime != null) {
-            executor.setKeepAliveTime(keepAliveTime, TimeUnit.SECONDS);
-        }
-    }
-}

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/listener/ThreadPoolRunListener.java

@@ -5,15 +5,15 @@ import io.dynamic.threadpool.common.config.ApplicationContextHolder;
 import io.dynamic.threadpool.common.constant.Constants;
 import io.dynamic.threadpool.common.model.PoolParameterInfo;
 import io.dynamic.threadpool.common.web.base.Result;
-import io.dynamic.threadpool.starter.config.DynamicThreadPoolBanner;
 import io.dynamic.threadpool.starter.config.CommonThreadPool;
+import io.dynamic.threadpool.starter.config.DynamicThreadPoolBanner;
 import io.dynamic.threadpool.starter.config.DynamicThreadPoolProperties;
 import io.dynamic.threadpool.starter.core.GlobalThreadPoolManage;
-import io.dynamic.threadpool.starter.toolkit.thread.QueueTypeEnum;
 import io.dynamic.threadpool.starter.remote.HttpAgent;
 import io.dynamic.threadpool.starter.remote.ServerHttpAgent;
+import io.dynamic.threadpool.starter.toolkit.thread.QueueTypeEnum;
+import io.dynamic.threadpool.starter.toolkit.thread.RejectedTypeEnum;
 import io.dynamic.threadpool.starter.toolkit.thread.ThreadPoolBuilder;
-import io.dynamic.threadpool.starter.wrap.CustomThreadPoolExecutor;
 import io.dynamic.threadpool.starter.wrap.DynamicThreadPoolWrap;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.annotation.Order;
@@ -68,7 +68,7 @@ public class ThreadPoolRunListener {
                             .keepAliveTime(ppi.getKeepAliveTime(), TimeUnit.SECONDS)
                             .workQueue(workQueue)
                             .threadFactory(tpId)
-                            .rejected(new CustomThreadPoolExecutor.AbortPolicy())
+                            .rejected(RejectedTypeEnum.createPolicy(ppi.getRejectedType()))
                             .build();
 
                     val.setPool(poolExecutor);