基于 Actuator 的可批改配置的线程池监控
1. 概要
之前公司因为应用线程池习惯不好，导致线程池负载负载过高。触发了回绝策略，导致大量工作失落。而并没有对这个状况进行监控，导致业务呈现故障之后才发现抛出了回绝异样。所以有必要对大量应用线程池的我的项目进行监控，并且最好能在不停机的状况下对线程池的参数进行批改，由此咱们能够用线程池的 hook 办法去对线程池的状态进行埋点，并且通过 Actuator 做可视化监控，自定义 Endpoint 去批改线程池外部参数，实现能够动静批改线程池参数。
2. 实现
1. 导入 Maven 依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>2. 编写 ThreadPoolMonitor.java 监控类
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 继承 ThreadPoolExecutor 类，笼罩了 shutdown(), shutdownNow(), beforeExecute() 和 afterExecute()
* 办法来统计线程池的执行状况
*/
public class ThreadPoolMonitor extends ThreadPoolExecutor {private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolMonitor.class);
/**
* 保留工作开始执行的工夫，当工作完结时，用工作完结工夫减去开始工夫计算工作执行工夫
*/
private final ConcurrentHashMap<String, Date> startTimes;
/**
* 线程池名称，个别以业务名称命名，不便辨别
*/
private final String poolName;
private long totalDiff;
/**
* 调用父类的构造方法，并初始化 HashMap 和线程池名称
*
* @param corePoolSize 线程池外围线程数
* @param maximumPoolSize 线程池最大线程数
* @param keepAliveTime 线程的最大闲暇工夫
* @param unit 闲暇工夫的单位
* @param workQueue 保留被提交工作的队列
* @param poolName 线程池名称
*/
public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue, String poolName) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
new EventThreadFactory(poolName), poolName);
}
/**
* 调用父类的构造方法，并初始化 HashMap 和线程池名称
*
* @param corePoolSize 线程池外围线程数
* @param maximumPoolSize 线程池最大线程数
* @param keepAliveTime 线程的最大闲暇工夫
* @param unit 闲暇工夫的单位
* @param workQueue 保留被提交工作的队列
* @param threadFactory 线程工厂
* @param poolName 线程池名称
*/
public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, String poolName) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
this.startTimes = new ConcurrentHashMap<>();
this.poolName = poolName;
}
/**
* 线程池提早敞开时（期待线程池里的工作都执行结束），统计线程池状况
*/
@Override
public void shutdown() {
// 统计已执行工作、正在执行工作、未执行工作数量
LOGGER.info("{} Going to shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",
this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
super.shutdown();}
/**
* 线程池立刻敞开时，统计线程池状况
*/
@Override
public List<Runnable> shutdownNow() {
// 统计已执行工作、正在执行工作、未执行工作数量
LOGGER.info("{} Going to immediately shutdown. Executed tasks: {}, Running tasks: {}, Pending tasks: {}",
this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
return super.shutdownNow();}
/**
* 工作执行之前，记录工作开始工夫
*/
@Override
protected void beforeExecute(Thread t, Runnable r) {startTimes.put(String.valueOf(r.hashCode()), new Date());
}
/**
* 工作执行之后，计算工作完结工夫
*/
@Override
protected void afterExecute(Runnable r, Throwable t) {Date startDate = startTimes.remove(String.valueOf(r.hashCode()));
Date finishDate = new Date();
long diff = finishDate.getTime() - startDate.getTime();
totalDiff += diff;
// 统计工作耗时、初始线程数、外围线程数、正在执行的工作数量、// 已实现工作数量、工作总数、队列里缓存的工作数量、池中存在的最大线程数、// 最大容许的线程数、线程闲暇工夫、线程池是否敞开、线程池是否终止
LOGGER.info("{}-pool-monitor:" +
"Duration: {} ms, PoolSize: {}, CorePoolSize: {}, Active: {}," +
"Completed: {}, Task: {}, Queue: {}, LargestPoolSize: {}," +
"MaximumPoolSize: {}, KeepAliveTime: {}, isShutdown: {}, isTerminated: {}",
this.poolName,
diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(),
this.getCompletedTaskCount(), this.getTaskCount(), this.getQueue().size(), this.getLargestPoolSize(),
this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS), this.isShutdown(), this.isTerminated());
}
/**
* 生成线程池所用的线程，只是改写了线程池默认的线程工厂，传入线程池名称，便于问题追踪
*/
static class EventThreadFactory implements ThreadFactory {private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
/**
* 初始化线程工厂
*
* @param poolName 线程池名称
*/
EventThreadFactory(String poolName) {SecurityManager s = System.getSecurityManager();
group = Objects.nonNull(s) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
namePrefix = poolName + "-pool-" + POOL_NUMBER.getAndIncrement() + "-thread-";}
@Override
public Thread newThread(Runnable r) {Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
if (t.isDaemon()) {t.setDaemon(false);
}
if (t.getPriority() != Thread.NORM_PRIORITY) {t.setPriority(Thread.NORM_PRIORITY);
}
return t;
}
}
public long getTotalDiff() {return totalDiff;}
}3. 实现 ResizeableBlockingQueue.java 可变队列
这里咱们间接批改 LinkedBlockingQueue 的代码，把 capacity 去掉 final，变成一个可变参数。再新增 get 和 set 办法。
/**
* The type Resizeable blocking queue.
*
* @param <E> the type parameter
*/
public class ResizeableBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private static final long serialVersionUID = -1232131234709194L;
/*
* 基于 LinkedBlockingQueue 实现的一个可变队列容量的阻塞队列
*
* */
/**
* The type Node.
*
* @param <E> the type parameter
*/
static class Node<E> {
E item;
Node<E> next;
Node(E x) {item = x;}
}
private int capacity;
private final AtomicInteger count = new AtomicInteger();
transient Node<E> head;
private transient Node<E> last;
private final ReentrantLock takeLock = new ReentrantLock();
private final Condition notEmpty = takeLock.newCondition();
private final ReentrantLock putLock = new ReentrantLock();
private final Condition notFull = putLock.newCondition();
/**
* Gets capacity.
*
* @return the capacity
*/
public int getCapacity() {return capacity;}
/**
* Sets capacity.
*
* @param capacity the capacity
*/
public void setCapacity(int capacity) {this.capacity = capacity;}
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {notEmpty.signal();
} finally {takeLock.unlock();
}
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {notFull.signal();
} finally {putLock.unlock();
}
}
private void enqueue(Node<E> node) {// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
last = last.next = node;
}
private E dequeue() {// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
E x = first.item;
first.item = null;
return x;
}
/**
* Fully lock.
*/
void fullyLock() {putLock.lock();
takeLock.lock();}
/**
* Fully unlock.
*/
void fullyUnlock() {takeLock.unlock();
putLock.unlock();}
/**
* Instantiates a new Resizeable blocking queue.
*/
public ResizeableBlockingQueue() {this(Integer.MAX_VALUE);
}
/**
* Instantiates a new Resizeable blocking queue.
*
* @param capacity the capacity
*/
public ResizeableBlockingQueue(int capacity) {if (capacity <= 0) {throw new IllegalArgumentException();
}
this.capacity = capacity;
last = head = new Node<E>(null);
}
/**
* Instantiates a new Resizeable blocking queue.
*
* @param c the c
*/
public ResizeableBlockingQueue(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 Node<E>(e));
++n;
}
count.set(n);
} finally {putLock.unlock();
}
}
// this doc comment is overridden to remove the reference to collections
// greater in size than Integer.MAX_VALUE
@Override
public int size() {return count.get();
}
// this doc comment is a modified copy of the inherited doc comment,
// without the reference to unlimited queues.
@Override
public int remainingCapacity() {return capacity - count.get();
}
@Override
public void put(E e) throws InterruptedException {if (e == null) {throw new NullPointerException();
}
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
/*
* Note that count is used in wait guard even though it is
* not protected by lock. This works because count can
* only decrease at this point (all other puts are shut
* out by lock), and we (or some other waiting put) are
* signalled if it ever changes from capacity. Similarly
* for all other uses of count in other wait guards.
*/
while (count.get() == capacity) {notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity) {notFull.signal();
}
} finally {putLock.unlock();
}
if (c == 0) {signalNotEmpty();
}
}
@Override
public boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException {if (e == null) {throw new NullPointerException();
}
long nanos = unit.toNanos(timeout);
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {while (count.get() == capacity) {if (nanos <= 0) {return false;}
nanos = notFull.awaitNanos(nanos);
}
enqueue(new Node<E>(e));
c = count.getAndIncrement();
if (c + 1 < capacity) {notFull.signal();
}
} finally {putLock.unlock();
}
if (c == 0) {signalNotEmpty();
}
return true;
}
@Override
public boolean offer(E e) {if (e == null) {throw new NullPointerException();
}
final AtomicInteger count = this.count;
if (count.get() == capacity) {return false;}
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {if (count.get() < capacity) {enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity) {notFull.signal();
}
}
} finally {putLock.unlock();
}
if (c == 0) {signalNotEmpty();
}
return c >= 0;
}
@Override
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {while (count.get() == 0) {notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1) {notEmpty.signal();
}
} finally {takeLock.unlock();
}
if (c == capacity) {signalNotFull();
}
return x;
}
@Override
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {while (count.get() == 0) {if (nanos <= 0) {return null;}
nanos = notEmpty.awaitNanos(nanos);
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1) {notEmpty.signal();
}
} finally {takeLock.unlock();
}
if (c == capacity) {signalNotFull();
}
return x;
}
@Override
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0) {return null;}
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {if (count.get() > 0) {x = dequeue();
c = count.getAndDecrement();
if (c > 1) {notEmpty.signal();
}
}
} finally {takeLock.unlock();
}
if (c == capacity) {signalNotFull();
}
return x;
}
@Override
public E peek() {if (count.get() == 0) {return null;}
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
Node<E> first = head.next;
if (first == null) {return null;} else {return first.item;}
} finally {takeLock.unlock();
}
}
void unlink(Node<E> p, Node<E> trail) {// assert isFullyLocked();
// p.next is not changed, to allow iterators that are
// traversing p to maintain their weak-consistency guarantee.
p.item = null;
trail.next = p.next;
if (last == p) {last = trail;}
if (count.getAndDecrement() == capacity) {notFull.signal();
}
}
@Override
public boolean remove(Object o) {if (o == null) {return false;}
fullyLock();
try {
for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) {if (o.equals(p.item)) {unlink(p, trail);
return true;
}
}
return false;
} finally {fullyUnlock();
}
}
@Override
public boolean contains(Object o) {if (o == null) {return false;}
fullyLock();
try {for (Node<E> p = head.next; p != null; p = p.next) {if (o.equals(p.item)) {return true;}
}
return false;
} finally {fullyUnlock();
}
}
@Override
public Object[] toArray() {fullyLock();
try {int size = count.get();
Object[] a = new Object[size];
int k = 0;
for (Node<E> p = head.next; p != null; p = p.next) {a[k++] = p.item;
}
return a;
} finally {fullyUnlock();
}
}
@Override
@SuppressWarnings("unchecked")
public <T> T[] toArray(T[] a) {fullyLock();
try {int size = count.get();
if (a.length < size) {a = (T[])java.lang.reflect.Array.newInstance
(a.getClass().getComponentType(), size);
}
int k = 0;
for (Node<E> p = head.next; p != null; p = p.next) {a[k++] = (T)p.item;
}
if (a.length > k) {a[k] = null;
}
return a;
} finally {fullyUnlock();
}
}
@Override
public String toString() {fullyLock();
try {
Node<E> p = head.next;
if (p == null) {return "[]";
}
StringBuilder sb = new StringBuilder();
sb.append('[');
for (;;) {
E e = p.item;
sb.append(e == this ? "(this Collection)" : e);
p = p.next;
if (p == null) {return sb.append(']').toString();}
sb.append(',').append(' ');
}
} finally {fullyUnlock();
}
}
@Override
public void clear() {fullyLock();
try {for (Node<E> p, h = head; (p = h.next) != null; h = p) {
h.next = h;
p.item = null;
}
head = last;
// assert head.item == null && head.next == null;
if (count.getAndSet(0) == capacity) {notFull.signal();
}
} finally {fullyUnlock();
}
}
@Override
public int drainTo(Collection<? super E> c) {return drainTo(c, Integer.MAX_VALUE);
}
@Override
public int drainTo(Collection<? super E> c, int maxElements) {if (c == null) {throw new NullPointerException();
}
if (c == this) {throw new IllegalArgumentException();
}
if (maxElements <= 0) {return 0;}
boolean signalNotFull = false;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {int n = Math.min(maxElements, count.get());
// count.get provides visibility to first n Nodes
Node<E> h = head;
int i = 0;
try {while (i < n) {
Node<E> p = h.next;
c.add(p.item);
p.item = null;
h.next = h;
h = p;
++i;
}
return n;
} finally {// Restore invariants even if c.add() threw
if (i > 0) {
// assert h.item == null;
head = h;
signalNotFull = (count.getAndAdd(-i) == capacity);
}
}
} finally {takeLock.unlock();
if (signalNotFull) {signalNotFull();
}
}
}
@Override
public Iterator<E> iterator() {return new Itr();
}
private class Itr implements Iterator<E> {
/*
* Basic weakly-consistent iterator. At all times hold the next
* item to hand out so that if hasNext() reports true, we will
* still have it to return even if lost race with a take etc.
*/
private Node<E> current;
private Node<E> lastRet;
private E currentElement;
Itr() {fullyLock();
try {
current = head.next;
if (current != null) {currentElement = current.item;}
} finally {fullyUnlock();
}
}
@Override
public boolean hasNext() {return current != null;}
private Node<E> nextNode(Node<E> p) {for (;;) {
Node<E> s = p.next;
if (s == p) {return head.next;}
if (s == null || s.item != null) {return s;}
p = s;
}
}
@Override
public E next() {fullyLock();
try {if (current == null) {throw new NoSuchElementException();
}
E x = currentElement;
lastRet = current;
current = nextNode(current);
currentElement = (current == null) ? null : current.item;
return x;
} finally {fullyUnlock();
}
}
@Override
public void remove() {if (lastRet == null) {throw new IllegalStateException();
}
fullyLock();
try {
Node<E> node = lastRet;
lastRet = null;
for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) {if (p == node) {unlink(p, trail);
break;
}
}
} finally {fullyUnlock();
}
}
}
/**
* The type Lbq spliterator.
*
* @param <E> the type parameter
*/
static final class LBQSpliterator<E> implements Spliterator<E> {
static final int MAX_BATCH = 1 << 25; // max batch array size;
final ResizeableBlockingQueue<E> queue;
Node<E> current; // current node; null until initialized
int batch; // batch size for splits
boolean exhausted; // true when no more nodes
long est; // size estimate
LBQSpliterator(ResizeableBlockingQueue<E> queue) {
this.queue = queue;
this.est = queue.size();}
@Override
public long estimateSize() { return est;}
@Override
public Spliterator<E> trySplit() {
Node<E> h;
final ResizeableBlockingQueue<E> q = this.queue;
int b = batch;
int n = (b <= 0) ? 1 : (b >= MAX_BATCH) ? MAX_BATCH : b + 1;
if (!exhausted &&
((h = current) != null || (h = q.head.next) != null) &&
h.next != null) {Object[] a = new Object[n];
int i = 0;
Node<E> p = current;
q.fullyLock();
try {if (p != null || (p = q.head.next) != null) {
do {if ((a[i] = p.item) != null) {++i;}
} while ((p = p.next) != null && i < n);
}
} finally {q.fullyUnlock();
}
if ((current = p) == null) {
est = 0L;
exhausted = true;
}
else if ((est -= i) < 0L) {est = 0L;}
if (i > 0) {
batch = i;
return Spliterators.spliterator
(a, 0, i, Spliterator.ORDERED | Spliterator.NONNULL |
Spliterator.CONCURRENT);
}
}
return null;
}
@Override
public void forEachRemaining(Consumer<? super E> action) {if (action == null) {throw new NullPointerException();
}
final ResizeableBlockingQueue<E> q = this.queue;
if (!exhausted) {
exhausted = true;
Node<E> p = current;
do {
E e = null;
q.fullyLock();
try {if (p == null) {p = q.head.next;}
while (p != null) {
e = p.item;
p = p.next;
if (e != null) {break;}
}
} finally {q.fullyUnlock();
}
if (e != null) {action.accept(e);
}
} while (p != null);
}
}
@Override
public boolean tryAdvance(Consumer<? super E> action) {if (action == null) {throw new NullPointerException();
}
final ResizeableBlockingQueue<E> q = this.queue;
if (!exhausted) {
E e = null;
q.fullyLock();
try {if (current == null) {current = q.head.next;}
while (current != null) {
e = current.item;
current = current.next;
if (e != null) {break;}
}
} finally {q.fullyUnlock();
}
if (current == null) {exhausted = true;}
if (e != null) {action.accept(e);
return true;
}
}
return false;
}
@Override
public int characteristics() {
return Spliterator.ORDERED | Spliterator.NONNULL |
Spliterator.CONCURRENT;
}
}
public Spliterator<E> spliterator() {return new LBQSpliterator<E>(this);
}
private void writeObject(java.io.ObjectOutputStream s)
throws java.io.IOException {fullyLock();
try {
// Write out any hidden stuff, plus capacity
s.defaultWriteObject();
// Write out all elements in the proper order.
for (Node<E> p = head.next; p != null; p = p.next) {s.writeObject(p.item);
}
// Use trailing null as sentinel
s.writeObject(null);
} finally {fullyUnlock();
}
}
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
// Read in capacity, and any hidden stuff
s.defaultReadObject();
count.set(0);
last = head = new Node<E>(null);
// Read in all elements and place in queue
for (;;) {@SuppressWarnings("unchecked")
E item = (E)s.readObject();
if (item == null) {break;}
add(item);
}
}
}
4. 实现 ThreadPoolUtil.java
编写线程池工具类，通过 Util 去创立线程池，并且用 HashMap 去指向创立的线程池，之后能够通过这个 HashMap 去获取线程池。
/**
* The type Thread pool util.
* 线程池工具类
*/
@Component
public class ThreadPoolUtil {
/**
* 通过 Hash 去指向创立的线程池，之后能够通过这个 HashMap 去获取线程池
*/
private final HashMap<String, ThreadPoolMonitor> threadPoolExecutorHashMap = new HashMap<>();
/**
* Creat thread pool thread pool monitor.
*
* 能够自定义队列类型的结构器
*
* @param corePoolSize the core pool size
* @param maximumPoolSize the maximum pool size
* @param keepAliveTime the keep alive time
* @param unit the unit
* @param workQueue the work queue
* @param poolName the pool name
* @return the thread pool monitor
*/
public ThreadPoolMonitor creatThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable> workQueue, String poolName) {ThreadPoolMonitor threadPoolExecutor = new ThreadPoolMonitor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, poolName);
threadPoolExecutorHashMap.put(poolName, threadPoolExecutor);
return threadPoolExecutor;
}
/**
* Creat thread pool thread pool monitor.
*
* ResizeableBlockingQueue 外面批改了 capacity 参数
* 能够通过 set 办法去批改队列的大小
* 应用默认队列的结构器
*
* @param corePoolSize the core pool size
* @param maximumPoolSize the maximum pool size
* @param keepAliveTime the keep alive time
* @param unit the unit
* @param queueSize the queue size
* @param poolName the pool name
* @return the thread pool monitor
*/
public ThreadPoolMonitor creatThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
TimeUnit unit, int queueSize, String poolName) {ThreadPoolMonitor threadPoolExecutor = new ThreadPoolMonitor(corePoolSize, maximumPoolSize, keepAliveTime, unit, new ResizeableBlockingQueue<>(queueSize), poolName);
threadPoolExecutorHashMap.put(poolName, threadPoolExecutor);
return threadPoolExecutor;
}
/**
* Gets thread pool executor hash map.
*
* @return the thread pool executor hash map
*/
public HashMap<String, ThreadPoolMonitor> getThreadPoolExecutorHashMap() {return threadPoolExecutorHashMap;}
}
5. 实现线程池信息的实体类
实现线程池信息的实体类用来 EndPoint 返回数据
ThreadPoolDetailInfo.java
/**
* The type Thread pool detail info.
*/
public class ThreadPoolDetailInfo {
private String threadPoolName;
private Integer poolSize;
private Integer corePoolSize;
private Integer largestPoolSize;
private Integer maximumPoolSize;
private long completedTaskCount;
private Integer active;
private long task;
private long keepAliveTime;
private String activePercent;
private Integer queueCapacity;
private Integer queueSize;
private long avgDiff;
/**
* Instantiates a new Thread pool detail info.
*
* @param threadPoolName the thread pool name
* @param poolSize the pool size
* @param corePoolSize the core pool size
* @param largestPoolSize the largest pool size
* @param maximumPoolSize the maximum pool size
* @param completedTaskCount the completed task count
* @param active the active
* @param task the task
* @param keepAliveTime the keep alive time
* @param activePercent the active percent
* @param queueCapacity the queue capacity
* @param queueSize the queue size
* @param avgDiff the avg diff
*/
public ThreadPoolDetailInfo(String threadPoolName, Integer poolSize, Integer corePoolSize, Integer largestPoolSize, Integer maximumPoolSize, long completedTaskCount, Integer active, long task, long keepAliveTime, String activePercent, Integer queueCapacity, Integer queueSize, long avgDiff) {
this.threadPoolName = threadPoolName;
this.poolSize = poolSize;
this.corePoolSize = corePoolSize;
this.largestPoolSize = largestPoolSize;
this.maximumPoolSize = maximumPoolSize;
this.completedTaskCount = completedTaskCount;
this.active = active;
this.task = task;
this.keepAliveTime = keepAliveTime;
this.activePercent = activePercent;
this.queueCapacity = queueCapacity;
this.queueSize = queueSize;
this.avgDiff = avgDiff;
}
/**
* Gets thread pool name.
*
* @return the thread pool name
*/
public String getThreadPoolName() {return threadPoolName;}
/**
* Sets thread pool name.
*
* @param threadPoolName the thread pool name
*/
public void setThreadPoolName(String threadPoolName) {this.threadPoolName = threadPoolName;}
/**
* Gets pool size.
*
* @return the pool size
*/
public Integer getPoolSize() {return poolSize;}
/**
* Sets pool size.
*
* @param poolSize the pool size
*/
public void setPoolSize(Integer poolSize) {this.poolSize = poolSize;}
/**
* Gets core pool size.
*
* @return the core pool size
*/
public Integer getCorePoolSize() {return corePoolSize;}
/**
* Sets core pool size.
*
* @param corePoolSize the core pool size
*/
public void setCorePoolSize(Integer corePoolSize) {this.corePoolSize = corePoolSize;}
/**
* Gets largest pool size.
*
* @return the largest pool size
*/
public Integer getLargestPoolSize() {return largestPoolSize;}
/**
* Sets largest pool size.
*
* @param largestPoolSize the largest pool size
*/
public void setLargestPoolSize(Integer largestPoolSize) {this.largestPoolSize = largestPoolSize;}
/**
* Gets maximum pool size.
*
* @return the maximum pool size
*/
public Integer getMaximumPoolSize() {return maximumPoolSize;}
/**
* Sets maximum pool size.
*
* @param maximumPoolSize the maximum pool size
*/
public void setMaximumPoolSize(Integer maximumPoolSize) {this.maximumPoolSize = maximumPoolSize;}
/**
* Gets completed task count.
*
* @return the completed task count
*/
public long getCompletedTaskCount() {return completedTaskCount;}
/**
* Sets completed task count.
*
* @param completedTaskCount the completed task count
*/
public void setCompletedTaskCount(long completedTaskCount) {this.completedTaskCount = completedTaskCount;}
/**
* Gets active.
*
* @return the active
*/
public Integer getActive() {return active;}
/**
* Sets active.
*
* @param active the active
*/
public void setActive(Integer active) {this.active = active;}
/**
* Gets task.
*
* @return the task
*/
public long getTask() {return task;}
/**
* Sets task.
*
* @param task the task
*/
public void setTask(long task) {this.task = task;}
/**
* Gets keep alive time.
*
* @return the keep alive time
*/
public long getKeepAliveTime() {return keepAliveTime;}
/**
* Sets keep alive time.
*
* @param keepAliveTime the keep alive time
*/
public void setKeepAliveTime(long keepAliveTime) {this.keepAliveTime = keepAliveTime;}
/**
* Gets active percent.
*
* @return the active percent
*/
public String getActivePercent() {return activePercent;}
/**
* Sets active percent.
*
* @param activePercent the active percent
*/
public void setActivePercent(String activePercent) {this.activePercent = activePercent;}
/**
* Gets queue capacity.
*
* @return the queue capacity
*/
public Integer getQueueCapacity() {return queueCapacity;}
/**
* Sets queue capacity.
*
* @param queueCapacity the queue capacity
*/
public void setQueueCapacity(Integer queueCapacity) {this.queueCapacity = queueCapacity;}
/**
* Gets queue size.
*
* @return the queue size
*/
public Integer getQueueSize() {return queueSize;}
/**
* Sets queue size.
*
* @param queueSize the queue size
*/
public void setQueueSize(Integer queueSize) {this.queueSize = queueSize;}
/**
* Gets avg diff.
*
* @return the avg diff
*/
public long getAvgDiff() {return avgDiff;}
/**
* Sets avg diff.
*
* @param avgDiff the avg diff
*/
public void setAvgDiff(long avgDiff) {this.avgDiff = avgDiff;}
}
ThreadPoolInfo.java
/**
* The type Thread pool info.
*/
public class ThreadPoolInfo {
private String threadPoolName;
private int corePoolSize;
private int maximumPoolSize;
private String queueType;
private int queueCapacity;
/**
* Instantiates a new Thread pool info.
*
* @param threadPoolName the thread pool name
* @param corePoolSize the core pool size
* @param maximumPoolSize the maximum pool size
* @param queueType the queue type
* @param queueCapacity the queue capacity
*/
public ThreadPoolInfo(String threadPoolName, int corePoolSize, int maximumPoolSize, String queueType, int queueCapacity) {
this.threadPoolName = threadPoolName;
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.queueType = queueType;
this.queueCapacity = queueCapacity;
}
/**
* Gets thread pool name.
*
* @return the thread pool name
*/
public String getThreadPoolName() {return threadPoolName;}
/**
* Sets thread pool name.
*
* @param threadPoolName the thread pool name
*/
public void setThreadPoolName(String threadPoolName) {this.threadPoolName = threadPoolName;}
/**
* Gets core pool size.
*
* @return the core pool size
*/
public int getCorePoolSize() {return corePoolSize;}
/**
* Sets core pool size.
*
* @param corePoolSize the core pool size
*/
public void setCorePoolSize(int corePoolSize) {this.corePoolSize = corePoolSize;}
/**
* Gets maximum pool size.
*
* @return the maximum pool size
*/
public int getMaximumPoolSize() {return maximumPoolSize;}
/**
* Sets maximum pool size.
*
* @param maximumPoolSize the maximum pool size
*/
public void setMaximumPoolSize(int maximumPoolSize) {this.maximumPoolSize = maximumPoolSize;}
/**
* Gets queue type.
*
* @return the queue type
*/
public String getQueueType() {return queueType;}
/**
* Sets queue type.
*
* @param queueType the queue type
*/
public void setQueueType(String queueType) {this.queueType = queueType;}
/**
* Gets capacity.
*
* @return the capacity
*/
public int getqueueCapacity() {return queueCapacity;}
/**
* Sets capacity.
*
* @param queueCapacity the queue capacity
*/
public void setqueueCapacity(int queueCapacity) {this.queueCapacity = queueCapacity;}
}
6. 编写 EndPoint
通过 actuator 里的 @RestControllerEndpoint 注解能够增加 Endpoints 接口。实质上是和 @Endpoint，@WebEndpoint 作用是一样的，都是为服务减少 actuator 接口，方便管理运行中的服务。然而有一个显著的不同是，@RestControllerEndpoint 只反对 Http 形式的拜访，不反对 JMX 的拜访。而且，端点的办法下面只反对 @GetMapping，@PostMapping，@DeleteMapping，@RequestMapping 等，而不反对 @ReadOperation，@WriteOperation，@DeleteOperation。而且它返回的格局是：application/json。
因为我司的监控零碎只反对 json 格局，实际上应用 Metrics 和 Grafana 去监控会更好。
/**
* The type Thread pool endpoint.
*
* @author newrank
*/
@RestControllerEndpoint(id = "threadpool")
@Component
public class ThreadPoolEndpoint {
@Autowired
private ThreadPoolUtil threadPoolUtil;
private static final ReentrantLock LOCK = new ReentrantLock();
private static final String RESIZEABLE_BLOCKING_QUEUE = "ResizeableBlockingQueue";
/**
* getThreadPools
* 获取以后所有线程池的线程名称
*/
@GetMapping("getThreadPools")
private List<String> getThreadPools (){List<String> threadPools = new ArrayList<>();
if (!threadPoolUtil.getThreadPoolExecutorHashMap().isEmpty()){for (Map.Entry<String, ThreadPoolMonitor> entry : threadPoolUtil.getThreadPoolExecutorHashMap().entrySet()) {threadPools.add(entry.getKey());
}
}
return threadPools;
}
/**
* 获取线程池可变参数信息
* @param threadPoolName
* @return
*/
@GetMapping("getThreadPoolFixInfo")
private ThreadPoolInfo getThreadPoolInfo(@RequestParam String threadPoolName){if (threadPoolUtil.getThreadPoolExecutorHashMap().containsKey(threadPoolName)){ThreadPoolMonitor threadPoolExecutor = threadPoolUtil.getThreadPoolExecutorHashMap().get(threadPoolName);
int queueCapacity = 0;
if (RESIZEABLE_BLOCKING_QUEUE.equals(threadPoolExecutor.getQueue().getClass().getSimpleName())){ResizeableBlockingQueue queue = (ResizeableBlockingQueue) threadPoolExecutor.getQueue();
queueCapacity = queue.getCapacity();}
return new ThreadPoolInfo(threadPoolName,threadPoolExecutor.getCorePoolSize(),threadPoolExecutor.getMaximumPoolSize(),
threadPoolExecutor.getQueue().getClass().getSimpleName(),queueCapacity);
}
return null;
}
/**
* 批改线程池配置
* @param threadPoolInfo
* @return
*/
@PostMapping("setThreadPoolFixInfo")
private Boolean setThreadPoolInfo(@RequestBody ThreadPoolInfo threadPoolInfo){if (threadPoolUtil.getThreadPoolExecutorHashMap().containsKey(threadPoolInfo.getThreadPoolName())){LOCK.lock();
try {ThreadPoolMonitor threadPoolExecutor = threadPoolUtil.getThreadPoolExecutorHashMap().get(threadPoolInfo.getThreadPoolName());
threadPoolExecutor.setMaximumPoolSize(threadPoolInfo.getMaximumPoolSize());
threadPoolExecutor.setCorePoolSize(threadPoolInfo.getCorePoolSize());
if (RESIZEABLE_BLOCKING_QUEUE.equals(threadPoolExecutor.getQueue().getClass().getSimpleName())){ResizeableBlockingQueue queue = (ResizeableBlockingQueue) threadPoolExecutor.getQueue();
queue.setCapacity(threadPoolInfo.getqueueCapacity());
}
return true;
}catch (Exception e){e.printStackTrace();
return false;
}
finally {LOCK.unlock();
}
}
return false;
}
/**
* 获取线程池监控信息
* @return
*/
@GetMapping("getThreadPoolListInfo")
private List<ThreadPoolDetailInfo> getThreadPoolListInfo(){List<ThreadPoolDetailInfo> detailInfoList = new ArrayList<>();
if (!threadPoolUtil.getThreadPoolExecutorHashMap().isEmpty()){for (Map.Entry<String, ThreadPoolMonitor> entry : threadPoolUtil.getThreadPoolExecutorHashMap().entrySet()) {ThreadPoolDetailInfo threadPoolDetailInfo = threadPoolInfo(entry.getValue(),entry.getKey());
detailInfoList.add(threadPoolDetailInfo);
}
}
return detailInfoList;
}
/**
* 组装线程池详情
* @param threadPool
* @param threadPoolName
* @return
*/
private ThreadPoolDetailInfo threadPoolInfo(ThreadPoolMonitor threadPool,String threadPoolName) {BigDecimal activeCount = new BigDecimal(threadPool.getActiveCount());
BigDecimal maximumPoolSize = new BigDecimal(threadPool.getMaximumPoolSize());
BigDecimal result =activeCount.divide(maximumPoolSize, 2, BigDecimal.ROUND_HALF_UP);
NumberFormat numberFormat = NumberFormat.getPercentInstance();
numberFormat.setMaximumFractionDigits(2);
int queueCapacity = 0;
if (RESIZEABLE_BLOCKING_QUEUE.equals(threadPool.getQueue().getClass().getSimpleName())){ResizeableBlockingQueue queue = (ResizeableBlockingQueue) threadPool.getQueue();
queueCapacity = queue.getCapacity();}
return new ThreadPoolDetailInfo(threadPoolName,threadPool.getPoolSize(), threadPool.getCorePoolSize(),
threadPool.getLargestPoolSize(), threadPool.getMaximumPoolSize(), threadPool.getCompletedTaskCount(),
threadPool.getActiveCount(),threadPool.getTaskCount(),threadPool.getKeepAliveTime(TimeUnit.MILLISECONDS),
numberFormat.format(result.doubleValue()),queueCapacity,threadPool.getQueue().size(),threadPool.getTotalDiff()/threadPool.getTaskCount());
}
}
7. 应用线程池监控
-
注解
@Async("asyncExecutor") public void getTrendQuery(){//do something} -
间接应用
public void test() {asyncExecutor.execute(()->{//do something} );
1. 查看线程详情
http://localhost/actuator/threadpool/getThreadPoolListInfo //GET 申请 返回：
[
{
"active": 0, // 正在进行的工作数
"activePercent": "0%",// 线程池负载
"completedTaskCount": 17, // 实现的工作数
"corePoolSize": 16, // 外围线程数
"keepAliveTime": 60000,// 线程存活工夫
"largestPoolSize": 16,// 达到的最大线程数
"maximumPoolSize": 32, // 最大线程数
"poolSize": 16,// 以后线程数
"queueCapacity": 500,// 队列长度 ps：如果不是 ResizeableBlockingQueue 队列则默认为 0
"task": 0, // 工作总数
"queueSize":0,// 队列中缓存的工作数量
"threadPoolName": "asyncExecutor" // 线程池名称
}
]2. 查看线程池参数
http://localhost/actuator/threadpool/getThreadPoolFixInfo?threadPoolName=asyncExecutor //GET 申请
参数:
| 名称 | 类型 |
|---|---|
| threadPoolName | String |
返回：
{
"corePoolSize": 16, // 外围线程数
"maximumPoolSize": 32, // 最大线程数
"queueCapacity": 500, // 队列大小
"queueType": "ResizeableBlockingQueue", // 队列类型
"threadPoolName": "asyncExecutor" // 线程池名称
}3. 批改线程池参数
https://localhost/actuator/th… //Post 申请
参数:
| 名称 | 类型 | 备注 |
|---|---|---|
| threadPoolName | String | |
| corePoolSize | int | 可变 |
| maximumPoolSize | int | 可变 |
| queueCapacity | int | 可变 |
| queueType | String | 不可变 |
申请类型：json
返回：Boolean
Github
Github
作者程度无限，若有谬误脱漏，请指出。
参考文章
1.Java 线程池实现原理及其在美团业务中的实际
2.Java 并发（六）线程池监控