序
本文主要研究一下Elasticsearch的SizeBlockingQueue
SizeBlockingQueue
elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/SizeBlockingQueue.java
public class SizeBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E> {
private final BlockingQueue<E> queue;
private final int capacity;
private final AtomicInteger size = new AtomicInteger();
public SizeBlockingQueue(BlockingQueue<E> queue, int capacity) {
assert capacity >= 0;
this.queue = queue;
this.capacity = capacity;
}
@Override
public int size() {
return size.get();
}
public int capacity() {
return this.capacity;
}
@Override
public Iterator<E> iterator() {
final Iterator<E> it = queue.iterator();
return new Iterator<E>() {
E current;
@Override
public boolean hasNext() {
return it.hasNext();
}
@Override
public E next() {
current = it.next();
return current;
}
@Override
public void remove() {
// note, we can't call #remove on the iterator because we need to know
// if it was removed or not
if (queue.remove(current)) {
size.decrementAndGet();
}
}
};
}
@Override
public E peek() {
return queue.peek();
}
@Override
public E poll() {
E e = queue.poll();
if (e != null) {
size.decrementAndGet();
}
return e;
}
@Override
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E e = queue.poll(timeout, unit);
if (e != null) {
size.decrementAndGet();
}
return e;
}
@Override
public boolean remove(Object o) {
boolean v = queue.remove(o);
if (v) {
size.decrementAndGet();
}
return v;
}
/**
* Forces adding an element to the queue, without doing size checks.
*/
public void forcePut(E e) throws InterruptedException {
size.incrementAndGet();
try {
queue.put(e);
} catch (InterruptedException ie) {
size.decrementAndGet();
throw ie;
}
}
@Override
public boolean offer(E e) {
while (true) {
final int current = size.get();
if (current >= capacity()) {
return false;
}
if (size.compareAndSet(current, 1 + current)) {
break;
}
}
boolean offered = queue.offer(e);
if (!offered) {
size.decrementAndGet();
}
return offered;
}
@Override
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {
// note, not used in ThreadPoolExecutor
throw new IllegalStateException("offer with timeout not allowed on size queue");
}
@Override
public void put(E e) throws InterruptedException {
// note, not used in ThreadPoolExecutor
throw new IllegalStateException("put not allowed on size queue");
}
@Override
public E take() throws InterruptedException {
E e;
try {
e = queue.take();
size.decrementAndGet();
} catch (InterruptedException ie) {
throw ie;
}
return e;
}
@Override
public int remainingCapacity() {
return capacity() - size.get();
}
@Override
public int drainTo(Collection<? super E> c) {
int v = queue.drainTo(c);
size.addAndGet(-v);
return v;
}
@Override
public int drainTo(Collection<? super E> c, int maxElements) {
int v = queue.drainTo(c, maxElements);
size.addAndGet(-v);
return v;
}
@Override
public Object[] toArray() {
return queue.toArray();
}
@Override
public <T> T[] toArray(T[] a) {
return (T[]) queue.toArray(a);
}
@Override
public boolean contains(Object o) {
return queue.contains(o);
}
@Override
public boolean containsAll(Collection<?> c) {
return queue.containsAll(c);
}
}
复制代码- SizeBlockingQueue继承了AbstractQueue，同时实现了BlockingQueue接口；它的构造器要求输入blockingQueue及capacity参数
- SizeBlockingQueue有个AtomicInteger类型的size参数用于记录queue的大小，它在poll、remove、offer、take等方法都会维护这个size参数
- 其中offer方法会判断当前size是否大于等于capacity，如果大于等于则直接返回false；而put方法则直接抛出IllegalStateException
ResizableBlockingQueue
elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/ResizableBlockingQueue.java
final class ResizableBlockingQueue<E> extends SizeBlockingQueue<E> {
private volatile int capacity;
ResizableBlockingQueue(BlockingQueue<E> queue, int initialCapacity) {
super(queue, initialCapacity);
this.capacity = initialCapacity;
}
@Override
public int capacity() {
return this.capacity;
}
@Override
public int remainingCapacity() {
return Math.max(0, this.capacity());
}
/** Resize the limit for the queue, returning the new size limit */
public synchronized int adjustCapacity(int optimalCapacity, int adjustmentAmount, int minCapacity, int maxCapacity) {
assert adjustmentAmount > 0 : "adjustment amount should be a positive value";
assert optimalCapacity >= 0 : "desired capacity cannot be negative";
assert minCapacity >= 0 : "cannot have min capacity smaller than 0";
assert maxCapacity >= minCapacity : "cannot have max capacity smaller than min capacity";
if (optimalCapacity == capacity) {
// Yahtzee!
return this.capacity;
}
if (optimalCapacity > capacity + adjustmentAmount) {
// adjust up
final int newCapacity = Math.min(maxCapacity, capacity + adjustmentAmount);
this.capacity = newCapacity;
return newCapacity;
} else if
(optimalCapacity < capacity - adjustmentAmount) {
// adjust down
final int newCapacity = Math.max(minCapacity, capacity - adjustmentAmount);
this.capacity = newCapacity;
return newCapacity;
} else {
return this.capacity;
}
}
}
复制代码- ResizableBlockingQueue继承了SizeBlockingQueue，它提供了一个线程安全的adjustCapacity方法，用于resize队列的capacity
EsExecutors
elasticsearch-7.0.1/server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java
public class EsExecutors {
//......
public static EsThreadPoolExecutor newFixed(String name, int size, int queueCapacity,
ThreadFactory threadFactory, ThreadContext contextHolder) {
BlockingQueue<Runnable> queue;
if (queueCapacity < 0) {
queue = ConcurrentCollections.newBlockingQueue();
} else {
queue = new SizeBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), queueCapacity);
}
return new EsThreadPoolExecutor(name, size, size, 0, TimeUnit.MILLISECONDS,
queue, threadFactory, new EsAbortPolicy(), contextHolder);
}
public static EsThreadPoolExecutor newAutoQueueFixed(String name, int size, int initialQueueCapacity, int minQueueSize,
int maxQueueSize, int frameSize, TimeValue targetedResponseTime,
ThreadFactory threadFactory, ThreadContext contextHolder) {
if (initialQueueCapacity <= 0) {
throw new IllegalArgumentException("initial queue capacity for [" + name + "] executor must be positive, got: " +
initialQueueCapacity);
}
ResizableBlockingQueue<Runnable> queue =
new ResizableBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), initialQueueCapacity);
return new QueueResizingEsThreadPoolExecutor(name, size, size, 0, TimeUnit.MILLISECONDS,
queue, minQueueSize, maxQueueSize, TimedRunnable::new, frameSize, targetedResponseTime, threadFactory,
new EsAbortPolicy(), contextHolder);
}
//......
}
复制代码- EsExecutors的newFixed创建的是使用SizeBlockingQueue的EsThreadPoolExecutor，而newAutoQueueFixed创建的是使用ResizableBlockingQueuede的QueueResizingEsThreadPoolExecutor
小结
- SizeBlockingQueue继承了AbstractQueue，同时实现了BlockingQueue接口；它的构造器要求输入blockingQueue及capacity参数；它有个AtomicInteger类型的size参数用于记录queue的大小，它在poll、remove、offer、take等方法都会维护这个size参数；其中offer方法会判断当前size是否大于等于capacity，如果大于等于则直接返回false；而put方法则直接抛出IllegalStateException
- ResizableBlockingQueue继承了SizeBlockingQueue，它提供了一个线程安全的adjustCapacity方法，用于resize队列的capacity
- EsExecutors的newFixed创建的是使用SizeBlockingQueue的EsThreadPoolExecutor，而newAutoQueueFixed创建的是使用ResizableBlockingQueuede的QueueResizingEsThreadPoolExecutor
