mirror of https://github.com/longtai-cn/hippo4j
parent
321d6f1eaf
commit
52dc1844dd
@ -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<T>`:
|
||||||
|
|
||||||
|
```java
|
||||||
|
public class MyArrayBlockingQueue implements CustomBlockingQueue<Runnable> {
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Integer getType() {
|
||||||
|
return 1001;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String getName() {
|
||||||
|
return "MyArrayBlockingQueue";
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public BlockingQueue<Runnable> 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<T> 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()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
|
||||||
@ -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
|
||||||
|
<T> BlockingQueue<T> of(Integer capacity) {
|
||||||
|
return new ResizableCapacityLinkedBlockingQueue<>(capacity);
|
||||||
|
}
|
||||||
|
@Override
|
||||||
|
<T> BlockingQueue<T> of() {
|
||||||
|
return new ResizableCapacityLinkedBlockingQueue<>();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
Creation and validation:
|
||||||
|
|
||||||
|
```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);
|
||||||
|
```
|
||||||
|
|
||||||
|
## 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".
|
||||||
|
|
||||||
Loading…
Reference in new issue