Allow passing queue capacity to custom blocking queue SPI

docs/docs/user_docs/dev_manual/queue-custom.md

@@ -24,48 +24,48 @@ public class MyArrayBlockingQueue implements CustomBlockingQueue<Runnable> {
     }
 
     @Override
-    public BlockingQueue<Runnable> generateBlockingQueue() {
-        return new ArrayBlockingQueue<>(256);
+    public BlockingQueue<Runnable> generateBlockingQueue(Integer capacity) {
+        int effectiveCapacity = capacity == null || capacity <= 0 ? 1024 : capacity;
+        return new ArrayBlockingQueue<>(effectiveCapacity);
     }
 }
 ```
 
+> 兼容提示：旧版只需实现 `generateBlockingQueue()` 的实现仍然有效，框架会在未覆写新方法时回退到旧逻辑，但推荐改为覆写带 `capacity` 入参的方法，以便直接复用服务端配置。
+
 ## 2. 声明 SPI 文件
 
 在 `src/main/resources/META-INF/services/` 目录下新增文件：
-
 ```
 cn.hippo4j.common.executor.support.CustomBlockingQueue
 ```
 
 文件内容仅一行：
-
 ```
 com.example.queue.MyArrayBlockingQueue
 ```
 
 ## 3. 服务端生效方式
 
-当服务端下发的 `queueType` 与 `capacity` 命中自定义类型时，框架会通过 SPI 自动创建队列。
+当服务端下发的 `queueType` 与 `capacity` 命中自定义类型时，框架会通过 SPI 自动创建队列，并将服务端配置的容量参数传入 `generateBlockingQueue(Integer capacity)`。
 
 ### 3.1 队列创建与验证
 
 ```java
-// 创建队列 - 使用 BlockingQueueTypeEnum
+// 创建队列
 BlockingQueue<T> q = BlockingQueueTypeEnum.createBlockingQueue(queueType, capacity);
+
 // 或者通过队列名称创建
 BlockingQueue<T> q2 = BlockingQueueTypeEnum.createBlockingQueue("ArrayBlockingQueue", capacity);
 
-// 验证队列配置 - 使用 BlockingQueueManager
+// 验证队列配置
 boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity);
 
-// 动态调整容量（仅 ResizableCapacityLinkedBlockingQueue 支持）- 使用 BlockingQueueManager
+// 动态调整容量（仅 ResizableCapacityLinkedBlockingQueue 支持）
 boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity);
 ```
 
-### 3.2 队列类型何时生效
-
-**重要说明**：队列类型（queueType）的变更需要客户端应用重启后生效。
+### 3.2 队列类型生效
 
 - **配置模板**：在线程池管理页面编辑队列类型，会保存到数据库，但不会推送到运行中的客户端。
 - **生效时机**：客户端应用重启时，会从服务端读取最新配置，并使用反射替换线程池的 `workQueue` 字段。
@@ -81,10 +81,9 @@ if (parameter.getCapacity() != null) {
         boolean success = BlockingQueueManager.changeQueueCapacity(
             executor.getQueue(), parameter.getCapacity());
         if (success) {
-            log.info("Queue capacity changed to: {}", parameter.getCapacity());
+            log.info("Queue capacity changed to: {} for thread pool: {}", 
+                     parameter.getCapacity(), parameter.getTpId());
         }
     }
 }
-```
-
-
+```

docs/docs/user_docs/dev_manual/queue-info.md

@@ -18,11 +18,9 @@ Hippo4j 内置多种常用阻塞队列类型，支持开箱即用，亦可通过
 - PriorityBlockingQueue（优先级队列）
 - ResizableCapacityLinkedBlockingQueue（可在线动态调容量的链表队列）
 
-其中 `ResizableCapacityLinkedBlockingQueue` 支持在线变更 `capacity`，无需重建线程池，适合动态调优场景。
+## 枚举定义
 
-## 代码对应
-
-枚举定义：
+**ResizableCapacityLinkedBlockingQueue**：支持在线变更 `capacity`。
 
 ```java
 // cn.hippo4j.common.executor.support.BlockingQueueTypeEnum
@@ -38,15 +36,6 @@ RESIZABLE_LINKED_BLOCKING_QUEUE(9, "ResizableCapacityLinkedBlockingQueue") {
 }
 ```
 
-创建与验证：
-
-```java
-// cn.hippo4j.common.executor.support.BlockingQueueManager
-BlockingQueue<T> q = BlockingQueueManager.createQueue(queueType, capacity);
-boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity);
-boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity);
-```
-
 ## 使用建议
 
 - 需要在线调容量：优先选择 `ResizableCapacityLinkedBlockingQueue`
@@ -55,6 +44,4 @@ boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCa
 - 需要优先级：选择 `PriorityBlockingQueue`
 - 需要同步移交：选择 `SynchronousQueue`
 
-如需自定义队列类型，请参考《阻塞队列自定义》。
-
-
+如需自定义队列类型，请参考《阻塞队列自定义》。

docs/i18n/zh/docusaurus-plugin-content-docs/current/user_docs/dev_manual/queue-custom.md

@@ -24,48 +24,48 @@ public class MyArrayBlockingQueue implements CustomBlockingQueue<Runnable> {
     }
 
     @Override
-    public BlockingQueue<Runnable> generateBlockingQueue() {
-        return new ArrayBlockingQueue<>(256);
+    public BlockingQueue<Runnable> generateBlockingQueue(Integer capacity) {
+        int effectiveCapacity = capacity == null || capacity <= 0 ? 1024 : capacity;
+        return new ArrayBlockingQueue<>(effectiveCapacity);
     }
 }
 ```
 
+> 兼容提示：旧版只需实现 `generateBlockingQueue()` 的实现仍然有效，框架会在未覆写新方法时回退到旧逻辑，但推荐改为覆写带 `capacity` 入参的方法，以便直接复用服务端配置。
+
 ## 2. 声明 SPI 文件
 
 在 `src/main/resources/META-INF/services/` 目录下新增文件：
-
 ```
 cn.hippo4j.common.executor.support.CustomBlockingQueue
 ```
 
 文件内容仅一行：
-
 ```
 com.example.queue.MyArrayBlockingQueue
 ```
 
 ## 3. 服务端生效方式
 
-当服务端下发的 `queueType` 与 `capacity` 命中自定义类型时，框架会通过 SPI 自动创建队列。
+当服务端下发的 `queueType` 与 `capacity` 命中自定义类型时，框架会通过 SPI 自动创建队列，并将服务端配置的容量参数传入 `generateBlockingQueue(Integer capacity)`。
 
 ### 3.1 队列创建与验证
 
 ```java
-// 创建队列 - 使用 BlockingQueueTypeEnum
+// 创建队列
 BlockingQueue<T> q = BlockingQueueTypeEnum.createBlockingQueue(queueType, capacity);
+
 // 或者通过队列名称创建
 BlockingQueue<T> q2 = BlockingQueueTypeEnum.createBlockingQueue("ArrayBlockingQueue", capacity);
 
-// 验证队列配置 - 使用 BlockingQueueManager
+// 验证队列配置
 boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity);
 
-// 动态调整容量（仅 ResizableCapacityLinkedBlockingQueue 支持）- 使用 BlockingQueueManager
+// 动态调整容量（仅 ResizableCapacityLinkedBlockingQueue 支持）
 boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity);
 ```
 
-### 3.2 队列类型何时生效
-
-**重要说明**：队列类型（queueType）的变更需要客户端应用重启后生效。
+### 3.2 队列类型生效
 
 - **配置模板**：在线程池管理页面编辑队列类型，会保存到数据库，但不会推送到运行中的客户端。
 - **生效时机**：客户端应用重启时，会从服务端读取最新配置，并使用反射替换线程池的 `workQueue` 字段。
@@ -81,10 +81,9 @@ if (parameter.getCapacity() != null) {
         boolean success = BlockingQueueManager.changeQueueCapacity(
             executor.getQueue(), parameter.getCapacity());
         if (success) {
-            log.info("Queue capacity changed to: {}", parameter.getCapacity());
+            log.info("Queue capacity changed to: {} for thread pool: {}", 
+                     parameter.getCapacity(), parameter.getTpId());
         }
     }
 }
-```
-
-
+```

docs/i18n/zh/docusaurus-plugin-content-docs/current/user_docs/dev_manual/queue-info.md

@@ -18,11 +18,9 @@ Hippo4j 内置多种常用阻塞队列类型，支持开箱即用，亦可通过
 - PriorityBlockingQueue（优先级队列）
 - ResizableCapacityLinkedBlockingQueue（可在线动态调容量的链表队列）
 
-其中 `ResizableCapacityLinkedBlockingQueue` 支持在线变更 `capacity`，无需重建线程池，适合动态调优场景。
+## 枚举定义
 
-## 代码对应
-
-枚举定义：
+**ResizableCapacityLinkedBlockingQueue**：支持在线变更 `capacity`。
 
 ```java
 // cn.hippo4j.common.executor.support.BlockingQueueTypeEnum
@@ -38,15 +36,6 @@ RESIZABLE_LINKED_BLOCKING_QUEUE(9, "ResizableCapacityLinkedBlockingQueue") {
 }
 ```
 
-创建与验证：
-
-```java
-// cn.hippo4j.common.executor.support.BlockingQueueManager
-BlockingQueue<T> q = BlockingQueueManager.createQueue(queueType, capacity);
-boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity);
-boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity);
-```
-
 ## 使用建议
 
 - 需要在线调容量：优先选择 `ResizableCapacityLinkedBlockingQueue`
@@ -55,6 +44,4 @@ boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCa
 - 需要优先级：选择 `PriorityBlockingQueue`
 - 需要同步移交：选择 `SynchronousQueue`
 
-如需自定义队列类型，请参考《阻塞队列自定义》。
-
-
+如需自定义队列类型，请参考《阻塞队列自定义》。

infra/common/src/main/java/cn/hippo4j/common/executor/support/BlockingQueueTypeEnum.java

@@ -238,17 +238,17 @@ public enum BlockingQueueTypeEnum {
         Collection<CustomBlockingQueue> customBlockingQueues = ServiceLoaderRegistry
                 .getSingletonServiceInstances(CustomBlockingQueue.class);
 
+        Integer resolvedCapacity = capacity;
+        if (resolvedCapacity == null || resolvedCapacity <= 0) {
+            resolvedCapacity = DEFAULT_CAPACITY;
+        }
+
+        Integer finalResolvedCapacity = resolvedCapacity;
         return customBlockingQueues.stream()
                 .filter(predicate)
-                .map(each -> each.generateBlockingQueue())
+                .map(each -> each.generateBlockingQueue(finalResolvedCapacity))
                 .findFirst()
-                .orElseGet(() -> {
-                    Integer tempCapacity = capacity;
-                    if (capacity == null || capacity <= 0) {
-                        tempCapacity = DEFAULT_CAPACITY;
-                    }
-                    return new LinkedBlockingQueue<T>(tempCapacity);
-                });
+                .orElseGet(() -> new LinkedBlockingQueue<T>(finalResolvedCapacity));
     }
 
     /**

infra/common/src/main/java/cn/hippo4j/common/executor/support/CustomBlockingQueue.java

@@ -42,8 +42,20 @@ public interface CustomBlockingQueue<T> {
 
     /**
      * Get custom blocking queue.
+     * Deprecated: override {@link #generateBlockingQueue(Integer)} to access capacity info.
      *
-     * @return
+     * @return blocking queue instance
      */
+    @Deprecated
     BlockingQueue<T> generateBlockingQueue();
+
+    /**
+     * Get custom blocking queue with capacity info from server configuration.
+     *
+     * @param capacity configured queue capacity (may be null or non-positive when not configured by user)
+     * @return blocking queue instance
+     */
+    default BlockingQueue<T> generateBlockingQueue(Integer capacity) {
+        return generateBlockingQueue();
+    }
 }

infra/common/src/test/java/cn/hippo4j/common/executor/integration/QueueSpiIntegrationTest.java

@@ -26,6 +26,7 @@ import org.junit.Test;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Queue SPI Integration Test: Verifies the full flow from parameters to queue creation
@@ -42,6 +43,8 @@ public class QueueSpiIntegrationTest {
      */
     public static class IntegrationTestQueue implements CustomBlockingQueue<Runnable> {
 
+        private static final AtomicInteger LAST_REQUESTED_CAPACITY = new AtomicInteger();
+
         @Override
         public Integer getType() {
             return 20001; // Integration test specific type ID
@@ -54,7 +57,22 @@ public class QueueSpiIntegrationTest {
 
         @Override
         public BlockingQueue<Runnable> generateBlockingQueue() {
-            return new ArrayBlockingQueue<>(256);
+            return generateBlockingQueue(256);
+        }
+
+        @Override
+        public BlockingQueue<Runnable> generateBlockingQueue(Integer capacity) {
+            int effectiveCapacity = capacity == null || capacity <= 0 ? 1024 : capacity;
+            LAST_REQUESTED_CAPACITY.set(effectiveCapacity);
+            return new ArrayBlockingQueue<>(effectiveCapacity);
+        }
+
+        public static int getLastRequestedCapacity() {
+            return LAST_REQUESTED_CAPACITY.get();
+        }
+
+        public static void resetLastRequestedCapacity() {
+            LAST_REQUESTED_CAPACITY.set(0);
         }
     }
 
@@ -112,10 +130,21 @@ public class QueueSpiIntegrationTest {
         Assert.assertNotNull("SPI custom queue should be created", spiQueue);
         Assert.assertTrue("Should be instance of ArrayBlockingQueue (TestCustomQueue implementation)",
                 spiQueue instanceof ArrayBlockingQueue);
+        Assert.assertEquals("Custom queue capacity should match server value", 512, spiQueue.remainingCapacity());
+
+        QueueSpiIntegrationTest.IntegrationTestQueue.resetLastRequestedCapacity();
+        BlockingQueue<Runnable> integrationQueue = BlockingQueueTypeEnum.createBlockingQueue(20001, 2048);
+        Assert.assertNotNull("Integration test custom queue should be created", integrationQueue);
+        Assert.assertTrue("Integration queue should be ArrayBlockingQueue", integrationQueue instanceof ArrayBlockingQueue);
+        Assert.assertEquals("Integration queue capacity should honor config", 2048, integrationQueue.remainingCapacity());
+        Assert.assertEquals("Integration custom queue should receive normalized capacity", 2048,
+                QueueSpiIntegrationTest.IntegrationTestQueue.getLastRequestedCapacity());
 
         System.out.println("SPI queue can be created via type ID");
         System.out.println("   Type 10001 (TestCustomQueue): " + spiQueue.getClass().getSimpleName());
         System.out.println("   Queue capacity: " + spiQueue.remainingCapacity());
+        System.out.println("   Type 20001 (IntegrationTestQueue): " + integrationQueue.getClass().getSimpleName());
+        System.out.println("   Queue capacity: " + integrationQueue.remainingCapacity());
     }
 
     /**
@@ -129,8 +158,8 @@ public class QueueSpiIntegrationTest {
         newParameter.setTenantId("default");
         newParameter.setItemId("item-001");
         newParameter.setTpId("test-pool");
-        newParameter.setQueueType(10001); // Switch to SPI custom queue
-        newParameter.setCapacity(512);
+        newParameter.setQueueType(20001); // Switch to IntegrationTestQueue
+        newParameter.setCapacity(768);
 
         System.out.println("Step 1: Config pushed - queueType=" + newParameter.getQueueType());
 
@@ -138,6 +167,7 @@ public class QueueSpiIntegrationTest {
         Assert.assertTrue("Should detect queue type change", queueTypeChanged);
         System.out.println("Step 2: Detected queue type change");
 
+        IntegrationTestQueue.resetLastRequestedCapacity();
         BlockingQueue<Runnable> newQueue = BlockingQueueTypeEnum.createBlockingQueue(
                 newParameter.getQueueType(),
                 newParameter.getCapacity());
@@ -146,8 +176,10 @@ public class QueueSpiIntegrationTest {
 
         Assert.assertTrue("New queue should be SPI custom implementation (ArrayBlockingQueue)",
                 newQueue instanceof ArrayBlockingQueue);
-        Assert.assertEquals("New queue capacity should be 512", 512, newQueue.remainingCapacity());
-        System.out.println("Step 4: Verified new queue - Type: ArrayBlockingQueue, Capacity: 512");
+        Assert.assertEquals("New queue capacity should be 768", 768, newQueue.remainingCapacity());
+        Assert.assertEquals("Custom queue should see normalized capacity", 768,
+                IntegrationTestQueue.getLastRequestedCapacity());
+        System.out.println("Step 4: Verified new queue - Type: ArrayBlockingQueue, Capacity: 768");
 
         System.out.println("Complete queue creation flow verified");
         System.out.println("Proves: Config → BlockingQueueTypeEnum → SPI → Custom Queue");
@@ -160,7 +192,7 @@ public class QueueSpiIntegrationTest {
     public void testQueueSwitchNotHardcoded() {
         System.out.println("\n========== Integration Test Scenario 5: Queue switch not hardcoded ==========");
 
-        int[] queueTypes = {1, 2, 3, 9, 10001};
+        int[] queueTypes = {1, 2, 3, 9, 10001, 20001};
         for (int queueType : queueTypes) {
             BlockingQueue<Runnable> queue = BlockingQueueTypeEnum.createBlockingQueue(queueType, 512);
             Assert.assertNotNull("Queue type " + queueType + " should be created", queue);
@@ -180,13 +212,16 @@ public class QueueSpiIntegrationTest {
 
         ThreadPoolParameterInfo parameter = new ThreadPoolParameterInfo();
         parameter.setRejectedType(1); // AbortPolicy
-        parameter.setQueueType(10001); // SPI custom queue
-        parameter.setCapacity(512);
+        parameter.setQueueType(20001); // SPI custom queue (IntegrationTestQueue)
+        parameter.setCapacity(640);
 
         BlockingQueue<Runnable> queue = BlockingQueueTypeEnum.createBlockingQueue(
                 parameter.getQueueType(),
                 parameter.getCapacity());
         Assert.assertNotNull("Queue should be created", queue);
+        Assert.assertEquals("Queue should reflect configured capacity", 640, queue.remainingCapacity());
+        Assert.assertEquals("Integration queue should receive normalized capacity", 640,
+                IntegrationTestQueue.getLastRequestedCapacity());
 
         System.out.println("Rejected policy and blocking queue design are consistent:");
         System.out.println("   - Rejected policy: created dynamically via type ID (rejectedType)");

infra/common/src/test/java/cn/hippo4j/common/executor/support/BlockingQueueSpiTest.java

@@ -23,6 +23,7 @@ import org.junit.Test;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Blocking Queue SPI Test: Verify that custom queues can be supported via SPI just like rejection policies
@@ -36,6 +37,10 @@ public class BlockingQueueSpiTest {
      */
     public static class TestCustomQueue implements CustomBlockingQueue<Runnable> {
 
+        private static final int DEFAULT_CAPACITY = 512;
+
+        private static final AtomicReference<Integer> LAST_REQUESTED_CAPACITY = new AtomicReference<>();
+
         @Override
         public Integer getType() {
             return 10001; // Custom type ID
@@ -48,10 +53,18 @@ public class BlockingQueueSpiTest {
 
         @Override
         public BlockingQueue<Runnable> generateBlockingQueue() {
-            // SPI implementation note: capacity should be passed from outside
-            // For simplicity in test, use fixed capacity 512
-            // In real project, capacity can be passed via constructor or other ways
-            return new ArrayBlockingQueue<>(512);
+            return generateBlockingQueue(DEFAULT_CAPACITY);
+        }
+
+        @Override
+        public BlockingQueue<Runnable> generateBlockingQueue(Integer capacity) {
+            int effectiveCapacity = capacity == null || capacity <= 0 ? DEFAULT_CAPACITY : capacity;
+            LAST_REQUESTED_CAPACITY.set(effectiveCapacity);
+            return new ArrayBlockingQueue<>(effectiveCapacity);
+        }
+
+        public static Integer getLastRequestedCapacity() {
+            return LAST_REQUESTED_CAPACITY.get();
         }
     }
 
@@ -100,6 +113,7 @@ public class BlockingQueueSpiTest {
         Assert.assertTrue("Should create ArrayBlockingQueue instance (TestCustomQueue implementation)",
                 spiQueue instanceof ArrayBlockingQueue);
         Assert.assertEquals("Queue capacity should be 512", 512, spiQueue.remainingCapacity());
+        Assert.assertEquals("Custom queue should receive requested capacity", Integer.valueOf(512), TestCustomQueue.getLastRequestedCapacity());
 
         System.out.println("Successfully created custom queue via SPI type ID 10001");
         System.out.println("Queue type: " + spiQueue.getClass().getSimpleName());
@@ -167,6 +181,23 @@ public class BlockingQueueSpiTest {
         System.out.println("Passed: BlockingQueueTypeEnum queue creation works");
     }
 
+    /**
+     * Test Case 3.5: SPI queue default capacity when config missing
+     */
+    @Test
+    public void testSpiQueueDefaultCapacity() {
+        System.out.println("\n========== Test Case 3.5: SPI queue default capacity ==========");
+
+        BlockingQueue<Runnable> queue = BlockingQueueTypeEnum.createBlockingQueue(10001, null);
+        Assert.assertNotNull("Should create custom queue when capacity missing", queue);
+        Assert.assertTrue("Should still be ArrayBlockingQueue", queue instanceof ArrayBlockingQueue);
+        Assert.assertEquals("Default capacity should fallback to 1024", 1024, queue.remainingCapacity());
+        Assert.assertEquals("Custom queue should receive normalized capacity", Integer.valueOf(1024), TestCustomQueue.getLastRequestedCapacity());
+
+        System.out.println("SPI queue default capacity -> " + queue.remainingCapacity());
+        System.out.println("Passed: Custom queue receives normalized capacity");
+    }
+
     /**
      * Test Case 4: Queue type recognition
      */

infra/common/src/test/resources/META-INF/services/cn.hippo4j.common.executor.support.CustomBlockingQueue

@@ -1 +1,2 @@
 cn.hippo4j.common.executor.support.BlockingQueueSpiTest$TestCustomQueue
+cn.hippo4j.common.executor.integration.QueueSpiIntegrationTest$IntegrationTestQueue