feat: 线程池阻塞队列 SPI 方式集成.

common/src/main/java/io/dynamic/threadpool/common/enums/QueueTypeEnum.java

@@ -1,52 +0,0 @@
-package io.dynamic.threadpool.common.enums;
-
-/**
- * 队列类型枚举
- *
- * @author chen.ma
- * @date 2021/6/25 12:30
- */
-public enum QueueTypeEnum {
-
-    /**
-     * {@link java.util.concurrent.ArrayBlockingQueue}
-     */
-    ARRAY_BLOCKING_QUEUE(1),
-
-    /**
-     * {@link java.util.concurrent.LinkedBlockingQueue}
-     */
-    LINKED_BLOCKING_QUEUE(2),
-
-    /**
-     * {@link java.util.concurrent.LinkedBlockingDeque}
-     */
-    LINKED_BLOCKING_DEQUE(3),
-
-    /**
-     * {@link java.util.concurrent.SynchronousQueue}
-     */
-    SYNCHRONOUS_QUEUE(4),
-
-    /**
-     * {@link java.util.concurrent.LinkedTransferQueue}
-     */
-    LINKED_TRANSFER_QUEUE(5),
-
-    /**
-     * {@link java.util.concurrent.PriorityBlockingQueue}
-     */
-    PRIORITY_BLOCKING_QUEUE(6),
-
-    /**
-     * {@link "io.dynamic.threadpool.starter.core.ResizableCapacityLinkedBlockIngQueue"}
-     */
-    RESIZABLE_LINKED_BLOCKING_QUEUE(9);
-
-    public Integer type;
-
-    QueueTypeEnum(int type) {
-        this.type = type;
-    }
-
-}

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/adapter/ThreadPoolConfigAdapter.java

@@ -1,7 +1,7 @@
 package io.dynamic.threadpool.starter.adapter;
 
 import io.dynamic.threadpool.common.config.ApplicationContextHolder;
-import io.dynamic.threadpool.common.enums.QueueTypeEnum;
+import io.dynamic.threadpool.starter.toolkit.thread.QueueTypeEnum;
 import io.dynamic.threadpool.starter.operation.ThreadPoolOperation;
 import io.dynamic.threadpool.starter.toolkit.thread.ThreadPoolBuilder;
 import io.dynamic.threadpool.starter.wrap.DynamicThreadPoolWrap;

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/common/CommonThreadPool.java

@@ -1,6 +1,6 @@
 package io.dynamic.threadpool.starter.common;
 
-import io.dynamic.threadpool.common.enums.QueueTypeEnum;
+import io.dynamic.threadpool.starter.toolkit.thread.QueueTypeEnum;
 import io.dynamic.threadpool.starter.toolkit.thread.ThreadPoolBuilder;
 import io.dynamic.threadpool.starter.toolkit.thread.RejectedPolicies;
 

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

@@ -1,6 +1,6 @@
 package io.dynamic.threadpool.starter.handler;
 
-import io.dynamic.threadpool.common.enums.QueueTypeEnum;
+import io.dynamic.threadpool.starter.toolkit.thread.QueueTypeEnum;
 import io.dynamic.threadpool.starter.core.ResizableCapacityLinkedBlockIngQueue;
 import lombok.extern.slf4j.Slf4j;
 

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

@@ -9,9 +9,9 @@ import io.dynamic.threadpool.starter.banner.DynamicThreadPoolBanner;
 import io.dynamic.threadpool.starter.common.CommonThreadPool;
 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.BlockingQueueUtil;
 import io.dynamic.threadpool.starter.toolkit.thread.ThreadPoolBuilder;
 import io.dynamic.threadpool.starter.wrap.CustomThreadPoolExecutor;
 import io.dynamic.threadpool.starter.wrap.DynamicThreadPoolWrap;
@@ -61,7 +61,7 @@ public class ThreadPoolRunListener {
                 result = httpAgent.httpGet(Constants.CONFIG_CONTROLLER_PATH, null, queryStrMap, 3000L);
                 if (result.isSuccess() && result.getData() != null && (ppi = JSON.toJavaObject((JSON) result.getData(), PoolParameterInfo.class)) != null) {
                     // 使用相关参数创建线程池
-                    BlockingQueue workQueue = BlockingQueueUtil.createBlockingQueue(ppi.getQueueType(), ppi.getCapacity());
+                    BlockingQueue workQueue = QueueTypeEnum.createBlockingQueue(ppi.getQueueType(), ppi.getCapacity());
                     ThreadPoolExecutor poolExecutor = ThreadPoolBuilder.builder()
                             .isCustomPool(true)
                             .poolThreadSize(ppi.getCoreSize(), ppi.getMaxSize())

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/spi/queue/CustomBlockingQueue.java

@@ -0,0 +1,27 @@
+package io.dynamic.threadpool.starter.spi.queue;
+
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * 自定义阻塞队列
+ *
+ * @author chen.ma
+ * @date 2021/7/11 00:51
+ */
+public interface CustomBlockingQueue {
+
+    /**
+     * 获取类型
+     *
+     * @return
+     */
+    Integer getType();
+
+    /**
+     * 生成阻塞队列
+     *
+     * @return
+     */
+    BlockingQueue generateBlockingQueue();
+
+}

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/toolkit/BlockingQueueUtil.java

@@ -1,38 +0,0 @@
-package io.dynamic.threadpool.starter.toolkit;
-
-import io.dynamic.threadpool.common.enums.QueueTypeEnum;
-import io.dynamic.threadpool.starter.core.ResizableCapacityLinkedBlockIngQueue;
-
-import java.util.Objects;
-import java.util.concurrent.*;
-
-/**
- * 阻塞队列工具类
- *
- * @author chen.ma
- * @date 2021/6/20 16:50
- */
-public class BlockingQueueUtil {
-
-    public static BlockingQueue createBlockingQueue(Integer type, Integer capacity) {
-        BlockingQueue blockingQueue = null;
-        if (Objects.equals(type, QueueTypeEnum.ARRAY_BLOCKING_QUEUE.type)) {
-            blockingQueue = new ArrayBlockingQueue(capacity);
-        } else if (Objects.equals(type, QueueTypeEnum.LINKED_BLOCKING_QUEUE.type)) {
-            blockingQueue = new LinkedBlockingQueue(capacity);
-        } else if (Objects.equals(type, QueueTypeEnum.LINKED_BLOCKING_DEQUE.type)) {
-            blockingQueue = new LinkedBlockingDeque(capacity);
-        } else if (Objects.equals(type, QueueTypeEnum.SYNCHRONOUS_QUEUE.type)) {
-            blockingQueue = new SynchronousQueue();
-        } else if (Objects.equals(type, QueueTypeEnum.LINKED_TRANSFER_QUEUE.type)) {
-            blockingQueue = new LinkedTransferQueue();
-        } else if (Objects.equals(type, QueueTypeEnum.PRIORITY_BLOCKING_QUEUE.type)) {
-            blockingQueue = new PriorityBlockingQueue(capacity);
-        } else if (Objects.equals(type, QueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.type)) {
-            blockingQueue = new ResizableCapacityLinkedBlockIngQueue(capacity);
-        } else {
-            throw new IllegalArgumentException("未找到类型匹配的阻塞队列.");
-        }
-        return blockingQueue;
-    }
-}

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/toolkit/thread/QueueTypeEnum.java

@@ -0,0 +1,93 @@
+package io.dynamic.threadpool.starter.toolkit.thread;
+
+import io.dynamic.threadpool.starter.core.ResizableCapacityLinkedBlockIngQueue;
+import io.dynamic.threadpool.starter.spi.DynamicTpServiceLoader;
+import io.dynamic.threadpool.starter.spi.queue.CustomBlockingQueue;
+
+import java.util.Collection;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.*;
+
+/**
+ * 队列类型枚举
+ *
+ * @author chen.ma
+ * @date 2021/6/25 12:30
+ */
+public enum QueueTypeEnum {
+
+    /**
+     * {@link java.util.concurrent.ArrayBlockingQueue}
+     */
+    ARRAY_BLOCKING_QUEUE(1),
+
+    /**
+     * {@link java.util.concurrent.LinkedBlockingQueue}
+     */
+    LINKED_BLOCKING_QUEUE(2),
+
+    /**
+     * {@link java.util.concurrent.LinkedBlockingDeque}
+     */
+    LINKED_BLOCKING_DEQUE(3),
+
+    /**
+     * {@link java.util.concurrent.SynchronousQueue}
+     */
+    SYNCHRONOUS_QUEUE(4),
+
+    /**
+     * {@link java.util.concurrent.LinkedTransferQueue}
+     */
+    LINKED_TRANSFER_QUEUE(5),
+
+    /**
+     * {@link java.util.concurrent.PriorityBlockingQueue}
+     */
+    PRIORITY_BLOCKING_QUEUE(6),
+
+    /**
+     * {@link "io.dynamic.threadpool.starter.core.ResizableCapacityLinkedBlockIngQueue"}
+     */
+    RESIZABLE_LINKED_BLOCKING_QUEUE(9);
+
+    public Integer type;
+
+    QueueTypeEnum(int type) {
+        this.type = type;
+    }
+
+    static {
+        DynamicTpServiceLoader.register(CustomBlockingQueue.class);
+    }
+
+    public static BlockingQueue createBlockingQueue(Integer type, Integer capacity) {
+        BlockingQueue blockingQueue = null;
+        if (Objects.equals(type, ARRAY_BLOCKING_QUEUE.type)) {
+            blockingQueue = new ArrayBlockingQueue(capacity);
+        } else if (Objects.equals(type, LINKED_BLOCKING_QUEUE.type)) {
+            blockingQueue = new LinkedBlockingQueue(capacity);
+        } else if (Objects.equals(type, LINKED_BLOCKING_DEQUE.type)) {
+            blockingQueue = new LinkedBlockingDeque(capacity);
+        } else if (Objects.equals(type, SYNCHRONOUS_QUEUE.type)) {
+            blockingQueue = new SynchronousQueue();
+        } else if (Objects.equals(type, LINKED_TRANSFER_QUEUE.type)) {
+            blockingQueue = new LinkedTransferQueue();
+        } else if (Objects.equals(type, PRIORITY_BLOCKING_QUEUE.type)) {
+            blockingQueue = new PriorityBlockingQueue(capacity);
+        } else if (Objects.equals(type, RESIZABLE_LINKED_BLOCKING_QUEUE.type)) {
+            blockingQueue = new ResizableCapacityLinkedBlockIngQueue(capacity);
+        }
+
+        Collection<CustomBlockingQueue> customBlockingQueues = DynamicTpServiceLoader
+                .getSingletonServiceInstances(CustomBlockingQueue.class);
+        blockingQueue = Optional.ofNullable(blockingQueue).orElseGet(() -> customBlockingQueues.stream()
+                .filter(each -> Objects.equals(type, each.getType()))
+                .map(each -> each.generateBlockingQueue())
+                .findFirst()
+                .orElse(new LinkedBlockingQueue(capacity)));
+
+        return blockingQueue;
+    }
+}

dynamic-threadpool-spring-boot-starter/src/main/java/io/dynamic/threadpool/starter/toolkit/thread/ThreadPoolBuilder.java

@@ -1,10 +1,8 @@
 package io.dynamic.threadpool.starter.toolkit.thread;
 
 
-import io.dynamic.threadpool.common.enums.QueueTypeEnum;
 import io.dynamic.threadpool.common.toolkit.Assert;
 import io.dynamic.threadpool.starter.builder.Builder;
-import io.dynamic.threadpool.starter.toolkit.BlockingQueueUtil;
 
 import java.math.BigDecimal;
 import java.util.concurrent.*;
@@ -239,7 +237,7 @@ public class ThreadPoolBuilder implements Builder<ThreadPoolExecutor> {
         // 快速消费线程池内置指定线程池
         if (!builder.isFastPool) {
             if (builder.workQueue == null) {
-                builder.workQueue = BlockingQueueUtil.createBlockingQueue(builder.queueType.type, builder.capacity);
+                builder.workQueue = QueueTypeEnum.createBlockingQueue(builder.queueType.type, builder.capacity);
             }
             initParam.setWorkQueue(builder.workQueue);
         }