springboot动态线程池、线程池监控

package com.itsoku.lesson009.comm; import java.util.*; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; /** * <b>description</b>： 支持扩容的阻塞队列 <br> * <b>time</b>：2024/4/2 23:39 <br> * <b>author</b>：ready */ public class ResizeLinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { private static final long serialVersionUID = -6903933977591709194L; /* * A variant of the "two lock queue" algorithm. The putLock gates * entry to put (and offer), and has an associated condition for * waiting puts. Similarly for the takeLock. The "count" field * that they both rely on is maintained as an atomic to avoid * needing to get both locks in most cases. Also, to minimize need * for puts to get takeLock and vice-versa, cascading notifies are * used. When a put notices that it has enabled at least one take, * it signals taker. That taker in turn signals others if more * items have been entered since the signal. And symmetrically for * takes signalling puts. Operations such as remove(Object) and * iterators acquire both locks. * * Visibility between writers and readers is provided as follows: * * Whenever an element is enqueued, the putLock is acquired and * count updated. A subsequent reader guarantees visibility to the * enqueued Node by either acquiring the putLock (via fullyLock) * or by acquiring the takeLock, and then reading n = count.get(); * this gives visibility to the first n items. * * To implement weakly consistent iterators, it appears we need to * keep all Nodes GC-reachable from a predecessor dequeued Node. * That would cause two problems: * - allow a rogue Iterator to cause unbounded memory retention * - cause cross-generational linking of old Nodes to new Nodes if * a Node was tenured while live, which generational GCs have a * hard time dealing with, causing repeated major collections. * However, only non-deleted Nodes need to be reachable from * dequeued Nodes, and reachability does not necessarily have to * be of the kind understood by the GC. We use the trick of * linking a Node that has just been dequeued to itself. Such a * self-link implicitly means to advance to head.next. */ /** * Linked list node class */ static class Node<E> { E item; /** * One of: * - the real successor Node * - this Node, meaning the successor is head.next * - null, meaning there is no successor (this is the last node) */ ResizeLinkedBlockingQueue.Node<E> next; Node(E x) { item = x; } } /** * The capacity bound, or Integer.MAX_VALUE if none */ private int capacity; /** * Current number of elements */ private final AtomicInteger count = new AtomicInteger(); /** * Head of linked list. * Invariant: head.item == null */ transient ResizeLinkedBlockingQueue.Node<E> head; /** * Tail of linked list. * Invariant: last.next == null */ private transient ResizeLinkedBlockingQueue.Node<E> last; /** * Lock held by take, poll, etc */ private final ReentrantLock takeLock = new ReentrantLock(); /** * Wait queue for waiting takes */ private final Condition notEmpty = takeLock.newCondition(); /** * Lock held by put, offer, etc */ private final ReentrantLock putLock = new ReentrantLock(); /** * Wait queue for waiting puts */ private final Condition notFull = putLock.newCondition(); /** * Signals a waiting take. Called only from put/offer (which do not * otherwise ordinarily lock takeLock.) */ private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); } } /** * Signals a waiting put. Called only from take/poll. */ private void signalNotFull() { final ReentrantLock putLock = this.putLock; putLock.lock(); try { notFull.signal(); } finally { putLock.unlock(); } } /** * Links node at end of queue. * * @param node the node */ private void enqueue(ResizeLinkedBlockingQueue.Node<E> node) { // assert putLock.isHeldByCurrentThread(); // assert last.next == null; last = last.next = node; } /** * Removes a node from head of queue. * * @return the node */ private E dequeue() { // assert takeLock.isHeldByCurrentThread(); // assert head.item == null; ResizeLinkedBlockingQueue.Node<E> h = head; ResizeLinkedBlockingQueue.Node<E> first = h.next; h.next = h; // help GC head = first; E x = first.item; first.item = null; return x; } /** * Locks to prevent both puts and takes. */ void fullyLock() { putLock.lock(); takeLock.lock(); } /** * Unlocks to allow both puts and takes. */ void fullyUnlock() { takeLock.unlock(); putLock.unlock(); } // /** // * Tells whether both locks are held by current thread. // */ // boolean isFullyLocked() { // return (putLock.isHeldByCurrentThread() && // takeLock.isHeldByCurrentThread()); // } /** * Creates a {@code ResizeLinkedBlockingQueue} with a capacity of * {@link Integer#MAX_VALUE}. */ public ResizeLinkedBlockingQueue() { this(Integer.MAX_VALUE); } /** * Creates a {@code ResizeLinkedBlockingQueue} with the given (fixed) capacity. * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity} is not greater * than zero */ public ResizeLinkedBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.capacity = capacity; last = head = new ResizeLinkedBlockingQueue.Node<E>(null); } /** * Creates a {@code ResizeLinkedBlockingQueue} with a capacity of * {@link Integer#MAX_VALUE}, initially containing the elements of the * given collection, * added in traversal order of the collection's iterator. * * @param c the collection of elements to initially contain * @throws NullPointerException if the specified collection or any * of its elements are null */ public ResizeLinkedBlockingQueue(Collection<? extends E> c) { this(Integer.MAX_VALUE); final ReentrantLock putLock = this.putLock; putLock.lock(); // Never contended, but necessary for visibility try { int n = 0; for (E e : c) { if (e == null) throw new NullPointerException(); if (n == capacity) throw new IllegalStateException("Queue full"); enqueue(new ResizeLinkedBlockingQueue.Node<E>(e)); ++n; } count.set(n); } finally { putLock.unlock(); } } /** * 设置容量 * @param capacity */ public void setCapacity(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); final ReentrantLock putLock = this.putLock; putLock.lock(); try { if (count.get() > capacity) { throw new IllegalArgumentExcepti