diff --git a/docs/docs/user_docs/dev_manual/queue-custom.md b/docs/docs/user_docs/dev_manual/queue-custom.md new file mode 100644 index 00000000..5d8c9075 --- /dev/null +++ b/docs/docs/user_docs/dev_manual/queue-custom.md @@ -0,0 +1,99 @@ +--- +sidebar_position: 4 +--- + +# Custom Blocking Queue + +Hippo4j extends blocking queues through SPI, allowing users to implement custom blocking queue types in Hippo4j. + +## 1. Define Custom Queue Class + +Implement the interface `cn.hippo4j.common.executor.support.CustomBlockingQueue`: + +```java +public class MyArrayBlockingQueue implements CustomBlockingQueue { + + @Override + public Integer getType() { + return 1001; + } + + @Override + public String getName() { + return "MyArrayBlockingQueue"; + } + + @Override + public BlockingQueue generateBlockingQueue() { + return new ArrayBlockingQueue<>(256); + } +} +``` + +## 2. Declare SPI File + +Create a file in the `src/main/resources/META-INF/services/` directory: + +``` +cn.hippo4j.common.executor.support.CustomBlockingQueue +``` + +File content (single line): + +``` +com.example.queue.MyArrayBlockingQueue +``` + +## 3. Server-side Activation + +When the `queueType` and `capacity` delivered by the server match the custom type, the framework will automatically create the queue through SPI. + +### 3.1 Queue Creation and Validation + +```java +// Create queue +BlockingQueue q = BlockingQueueManager.createQueue(queueType, capacity); + +// Validate queue configuration +boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity); + +// Dynamic capacity adjustment (only supported by ResizableCapacityLinkedBlockingQueue) +boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity); +``` + +### 3.2 Queue Type Switching + +When you need to switch queue types, use the `ThreadPoolRebuilder.rebuildAndSwitch` method, which creates a new thread pool instance and safely migrates tasks: + +```java +boolean ok = ThreadPoolRebuilder.rebuildAndSwitch( + executor, // Current thread pool + newQueueType, // New queue type + capacity, // Queue capacity + threadPoolId // Thread pool ID +); +``` + +Server-side dynamic refresh implementation: + +```java +// ServerThreadPoolDynamicRefresh#handleQueueChanges +boolean queueTypeChanged = parameter.getQueueType() != null && + !Objects.equals(BlockingQueueManager.getQueueType(executor.getQueue()), parameter.getQueueType()); + +if (queueTypeChanged) { + // Use safe rebuild approach for queue switching + boolean ok = ThreadPoolRebuilder.rebuildAndSwitch( + executor, + parameter.getQueueType(), + parameter.getCapacity(), + threadPoolId + ); + if (ok) { + log.info("Queue type rebuilt and switched to: {}", + BlockingQueueTypeEnum.getBlockingQueueNameByType(parameter.getQueueType())); + } +} +``` + + diff --git a/docs/docs/user_docs/dev_manual/queue-info.md b/docs/docs/user_docs/dev_manual/queue-info.md new file mode 100644 index 00000000..cd28fa8d --- /dev/null +++ b/docs/docs/user_docs/dev_manual/queue-info.md @@ -0,0 +1,59 @@ +--- +sidebar_position: 3 +--- + +# Built-in Blocking Queues + +Hippo4j provides multiple built-in blocking queue types that are ready to use out of the box. You can also extend custom queue types through SPI. + +## Built-in Queue Types + +The following types can be directly selected in the server or configuration (Enum: `BlockingQueueTypeEnum`): + +- ArrayBlockingQueue (bounded array-based queue) +- LinkedBlockingQueue (linked list queue) +- LinkedBlockingDeque (double-ended queue) +- SynchronousQueue (synchronous handoff queue) +- LinkedTransferQueue (transferable queue) +- PriorityBlockingQueue (priority queue) +- ResizableCapacityLinkedBlockingQueue (dynamically resizable linked list queue) + +Among them, `ResizableCapacityLinkedBlockingQueue` supports online capacity changes without rebuilding the thread pool, making it suitable for dynamic tuning scenarios. + +## Code Reference + +Enum definition: + +```java +// cn.hippo4j.common.executor.support.BlockingQueueTypeEnum +RESIZABLE_LINKED_BLOCKING_QUEUE(9, "ResizableCapacityLinkedBlockingQueue") { + @Override + BlockingQueue of(Integer capacity) { + return new ResizableCapacityLinkedBlockingQueue<>(capacity); + } + @Override + BlockingQueue of() { + return new ResizableCapacityLinkedBlockingQueue<>(); + } +} +``` + +Creation and validation: + +```java +// cn.hippo4j.common.executor.support.BlockingQueueManager +BlockingQueue q = BlockingQueueManager.createQueue(queueType, capacity); +boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity); +boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity); +``` + +## Usage Recommendations + +- Need online capacity adjustment: prioritize `ResizableCapacityLinkedBlockingQueue` +- Need strictly bounded: choose `ArrayBlockingQueue` +- Need unbounded throughput: choose `LinkedBlockingQueue` +- Need priority: choose `PriorityBlockingQueue` +- Need synchronous handoff: choose `SynchronousQueue` + +For custom queue types, please refer to "Custom Blocking Queue". + diff --git a/docs/i18n/zh/docusaurus-plugin-content-docs/current/user_docs/dev_manual/queue-custom.md b/docs/i18n/zh/docusaurus-plugin-content-docs/current/user_docs/dev_manual/queue-custom.md index 2962c683..addfbfcf 100644 --- a/docs/i18n/zh/docusaurus-plugin-content-docs/current/user_docs/dev_manual/queue-custom.md +++ b/docs/i18n/zh/docusaurus-plugin-content-docs/current/user_docs/dev_manual/queue-custom.md @@ -11,12 +11,6 @@ Hippo4j 通过 SPI 的方式对拒绝策略进行扩展，可以让用户在 Hip 实现接口 `cn.hippo4j.common.executor.support.CustomBlockingQueue`： ```java -package com.example.queue; - -import cn.hippo4j.common.executor.support.CustomBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ArrayBlockingQueue; - public class MyArrayBlockingQueue implements CustomBlockingQueue { @Override @@ -52,21 +46,35 @@ com.example.queue.MyArrayBlockingQueue ## 3. 服务端生效方式 -当服务端下发的 `queueType` 与 `capacity` 命中自定义类型时，框架会通过 SPI 自动创建队列： +当服务端下发的 `queueType` 与 `capacity` 命中自定义类型时，框架会通过 SPI 自动创建队列。 + +### 3.1 队列创建与验证 ```java -// 创建与验证 +// 创建队列 BlockingQueue q = BlockingQueueManager.createQueue(queueType, capacity); -boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity); -// 在线替换 -boolean swapped = BlockingQueueManager.replaceQueue(executor, q); +// 验证队列配置 +boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity); // 动态调整容量（仅 ResizableCapacityLinkedBlockingQueue 支持） boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity); ``` -服务端动态刷新处： +### 3.2 队列类型切换 + +当需要切换队列类型时，使用 `ThreadPoolRebuilder.rebuildAndSwitch` 方法，该方法会创建新的线程池实例并安全地迁移任务： + +```java +boolean ok = ThreadPoolRebuilder.rebuildAndSwitch( + executor, // 当前线程池 + newQueueType, // 新队列类型 + capacity, // 队列容量 + threadPoolId // 线程池ID +); +``` + +服务端动态刷新处的实现： ```java // ServerThreadPoolDynamicRefresh#handleQueueChanges @@ -74,9 +82,17 @@ boolean queueTypeChanged = parameter.getQueueType() != null && !Objects.equals(BlockingQueueManager.getQueueType(executor.getQueue()), parameter.getQueueType()); if (queueTypeChanged) { - boolean swapped = BlockingQueueManager.replaceQueue( - executor, BlockingQueueManager.createQueue(parameter.getQueueType(), parameter.getCapacity())); - ... + // 使用安全的重建方式切换队列 + boolean ok = ThreadPoolRebuilder.rebuildAndSwitch( + executor, + parameter.getQueueType(), + parameter.getCapacity(), + threadPoolId + ); + if (ok) { + log.info("Queue type rebuilt and switched to: {}", + BlockingQueueTypeEnum.getBlockingQueueNameByType(parameter.getQueueType())); + } } ``` diff --git a/docs/i18n/zh/docusaurus-plugin-content-docs/version-1.5.0/user_docs/dev_manual/queue-custom.md b/docs/i18n/zh/docusaurus-plugin-content-docs/version-1.5.0/user_docs/dev_manual/queue-custom.md deleted file mode 100644 index 2962c683..00000000 --- a/docs/i18n/zh/docusaurus-plugin-content-docs/version-1.5.0/user_docs/dev_manual/queue-custom.md +++ /dev/null @@ -1,83 +0,0 @@ ---- -sidebar_position: 4 ---- - -# 阻塞队列自定义 - -Hippo4j 通过 SPI 的方式对拒绝策略进行扩展，可以让用户在 Hippo4j 中完成自定义阻塞队列实现。 - -## 1. 定义自定义队列类 - -实现接口 `cn.hippo4j.common.executor.support.CustomBlockingQueue`： - -```java -package com.example.queue; - -import cn.hippo4j.common.executor.support.CustomBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ArrayBlockingQueue; - -public class MyArrayBlockingQueue implements CustomBlockingQueue { - - @Override - public Integer getType() { - return 1001; - } - - @Override - public String getName() { - return "MyArrayBlockingQueue"; - } - - @Override - public BlockingQueue generateBlockingQueue() { - return new ArrayBlockingQueue<>(256); - } -} -``` - -## 2. 声明 SPI 文件 - -在 `src/main/resources/META-INF/services/` 目录下新增文件： - -``` -cn.hippo4j.common.executor.support.CustomBlockingQueue -``` - -文件内容仅一行： - -``` -com.example.queue.MyArrayBlockingQueue -``` - -## 3. 服务端生效方式 - -当服务端下发的 `queueType` 与 `capacity` 命中自定义类型时，框架会通过 SPI 自动创建队列： - -```java -// 创建与验证 -BlockingQueue q = BlockingQueueManager.createQueue(queueType, capacity); -boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity); - -// 在线替换 -boolean swapped = BlockingQueueManager.replaceQueue(executor, q); - -// 动态调整容量（仅 ResizableCapacityLinkedBlockingQueue 支持） -boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity); -``` - -服务端动态刷新处： - -```java -// ServerThreadPoolDynamicRefresh#handleQueueChanges -boolean queueTypeChanged = parameter.getQueueType() != null && - !Objects.equals(BlockingQueueManager.getQueueType(executor.getQueue()), parameter.getQueueType()); - -if (queueTypeChanged) { - boolean swapped = BlockingQueueManager.replaceQueue( - executor, BlockingQueueManager.createQueue(parameter.getQueueType(), parameter.getCapacity())); - ... -} -``` - - diff --git a/docs/i18n/zh/docusaurus-plugin-content-docs/version-1.5.0/user_docs/dev_manual/queue-info.md b/docs/i18n/zh/docusaurus-plugin-content-docs/version-1.5.0/user_docs/dev_manual/queue-info.md deleted file mode 100644 index 47e372db..00000000 --- a/docs/i18n/zh/docusaurus-plugin-content-docs/version-1.5.0/user_docs/dev_manual/queue-info.md +++ /dev/null @@ -1,60 +0,0 @@ ---- -sidebar_position: 3 ---- - -# 内置阻塞队列 - -Hippo4j 内置多种常用阻塞队列类型，支持开箱即用，亦可通过 SPI 扩展自定义队列类型。 - -## 内置类型清单 - -以下类型可直接在服务端或配置中选择（枚举：`BlockingQueueTypeEnum`）： - -- ArrayBlockingQueue（数组有界队列） -- LinkedBlockingQueue（链表队列） -- LinkedBlockingDeque（双端队列） -- SynchronousQueue（同步移交队列） -- LinkedTransferQueue（可转移队列） -- PriorityBlockingQueue（优先级队列） -- ResizableCapacityLinkedBlockingQueue（可在线动态调容量的链表队列） - -其中 `ResizableCapacityLinkedBlockingQueue` 支持在线变更 `capacity`，无需重建线程池，适合动态调优场景。 - -## 代码对应 - -枚举定义： - -```java -// cn.hippo4j.common.executor.support.BlockingQueueTypeEnum -RESIZABLE_LINKED_BLOCKING_QUEUE(9, "ResizableCapacityLinkedBlockingQueue") { - @Override - BlockingQueue of(Integer capacity) { - return new ResizableCapacityLinkedBlockingQueue<>(capacity); - } - @Override - BlockingQueue of() { - return new ResizableCapacityLinkedBlockingQueue<>(); - } -} -``` - -创建与验证： - -```java -// cn.hippo4j.common.executor.support.BlockingQueueManager -BlockingQueue q = BlockingQueueManager.createQueue(queueType, capacity); -boolean valid = BlockingQueueManager.validateQueueConfig(queueType, capacity); -boolean ok = BlockingQueueManager.changeQueueCapacity(executor.getQueue(), newCapacity); -``` - -## 使用建议 - -- 需要在线调容量：优先选择 `ResizableCapacityLinkedBlockingQueue` -- 需要严格有界：选择 `ArrayBlockingQueue` -- 需要无界吞吐：选择 `LinkedBlockingQueue` -- 需要优先级：选择 `PriorityBlockingQueue` -- 需要同步移交：选择 `SynchronousQueue` - -如需自定义队列类型，请参考《阻塞队列自定义》。 - - diff --git a/infra/common/src/main/java/cn/hippo4j/common/executor/support/BlockingQueueManager.java b/infra/common/src/main/java/cn/hippo4j/common/executor/support/BlockingQueueManager.java index c58851b2..2b4138c8 100644 --- a/infra/common/src/main/java/cn/hippo4j/common/executor/support/BlockingQueueManager.java +++ b/infra/common/src/main/java/cn/hippo4j/common/executor/support/BlockingQueueManager.java @@ -126,37 +126,6 @@ public class BlockingQueueManager { return false; } - /** - * Replace queue in thread pool executor - * - * @param executor thread pool executor - * @param newQueue new queue instance - * @param queue element type - * @return true if queue was replaced - */ - public static boolean replaceQueue(ThreadPoolExecutor executor, BlockingQueue newQueue) { - if (executor == null || newQueue == null) { - return false; - } - try { - if (executor.getActiveCount() > 0 || !executor.getQueue().isEmpty()) { - return false; - } - try { - java.lang.reflect.Field field = ThreadPoolExecutor.class.getDeclaredField("workQueue"); - field.setAccessible(true); - field.set(executor, newQueue); - return true; - } catch (Throwable ignore) { - log.warn("JDK security prevents replacing workQueue; skip."); - return false; - } - } catch (Exception e) { - log.error("Failed to replace queue", e); - return false; - } - } - /** * Get queue type from queue instance *