mirror of https://github.com/longtai-cn/hippo4j
chen.ma
4 years ago
parent
847f87f790
commit
5383861e2f
@ -0,0 +1,20 @@
|
||||
package io.dynamic.threadpool.starter.builder;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* 建造者模式接口定义
|
||||
*
|
||||
* @author chen.ma
|
||||
* @date 2021/7/5 21:39
|
||||
*/
|
||||
public interface Builder<T> extends Serializable {
|
||||
|
||||
/**
|
||||
* 构建
|
||||
*
|
||||
* @return 被构建的对象
|
||||
*/
|
||||
T build();
|
||||
|
||||
}
|
||||
@ -0,0 +1,41 @@
|
||||
package io.dynamic.threadpool.starter.toolkit;
|
||||
|
||||
import java.lang.reflect.Array;
|
||||
|
||||
/**
|
||||
* Array Util.
|
||||
*
|
||||
* @author chen.ma
|
||||
* @date 2021/7/5 21:54
|
||||
*/
|
||||
public class ArrayUtil {
|
||||
|
||||
public static <T> T[] addAll(final T[] array1, @SuppressWarnings("unchecked") final T... array2) {
|
||||
if (array1 == null) {
|
||||
return clone(array2);
|
||||
} else if (array2 == null) {
|
||||
return clone(array1);
|
||||
}
|
||||
final Class<?> type1 = array1.getClass().getComponentType();
|
||||
@SuppressWarnings("unchecked") final T[] joinedArray = (T[]) Array.newInstance(type1, array1.length + array2.length);
|
||||
System.arraycopy(array1, 0, joinedArray, 0, array1.length);
|
||||
try {
|
||||
System.arraycopy(array2, 0, joinedArray, array1.length, array2.length);
|
||||
} catch (final ArrayStoreException ase) {
|
||||
final Class<?> type2 = array2.getClass().getComponentType();
|
||||
if (!type1.isAssignableFrom(type2)) {
|
||||
throw new IllegalArgumentException("Cannot store " + type2.getName() + " in an array of "
|
||||
+ type1.getName(), ase);
|
||||
}
|
||||
throw ase;
|
||||
}
|
||||
return joinedArray;
|
||||
}
|
||||
|
||||
public static <T> T[] clone(final T[] array) {
|
||||
if (array == null) {
|
||||
return null;
|
||||
}
|
||||
return array.clone();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,140 @@
|
||||
package io.dynamic.threadpool.starter.toolkit.thread;
|
||||
|
||||
import io.dynamic.threadpool.common.toolkit.Assert;
|
||||
import lombok.Data;
|
||||
import lombok.experimental.Accessors;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.concurrent.*;
|
||||
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
|
||||
|
||||
/**
|
||||
* 抽象线程池模版构建
|
||||
*
|
||||
* @author chen.ma
|
||||
* @date 2021/7/5 21:45
|
||||
*/
|
||||
@Slf4j
|
||||
public class AbstractBuildThreadPoolTemplate {
|
||||
|
||||
/**
|
||||
* 线程池构建初始化参数
|
||||
* <p>
|
||||
* 此处本身是模版设计方法, 但是考虑创建简洁性, 移除 abstract
|
||||
* 异常参考 {@link AbstractQueuedSynchronizer#tryAcquire}
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
protected static ThreadPoolInitParam initParam() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建线程池
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static ThreadPoolExecutor buildPool() {
|
||||
ThreadPoolInitParam initParam = initParam();
|
||||
return buildPool(initParam);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建线程池
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static ThreadPoolExecutor buildPool(ThreadPoolInitParam initParam) {
|
||||
Assert.notNull(initParam);
|
||||
ThreadPoolExecutor executorService =
|
||||
new ThreadPoolExecutorTemplate(initParam.getCorePoolNum(),
|
||||
initParam.getMaxPoolNum(),
|
||||
initParam.getKeepAliveTime(),
|
||||
initParam.getTimeUnit(),
|
||||
initParam.getWorkQueue(),
|
||||
initParam.getThreadFactory(),
|
||||
initParam.rejectedExecutionHandler);
|
||||
return executorService;
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建快速执行线程池
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static ThreadPoolExecutor buildFastPool() {
|
||||
ThreadPoolInitParam initParam = initParam();
|
||||
return buildFastPool(initParam);
|
||||
}
|
||||
|
||||
/**
|
||||
* 构建快速执行线程池
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static ThreadPoolExecutor buildFastPool(ThreadPoolInitParam initParam) {
|
||||
TaskQueue<Runnable> taskQueue = new TaskQueue(initParam.getCapacity());
|
||||
FastThreadPoolExecutor fastThreadPoolExecutor =
|
||||
new FastThreadPoolExecutor(initParam.getCorePoolNum(),
|
||||
initParam.getMaxPoolNum(),
|
||||
initParam.getKeepAliveTime(),
|
||||
initParam.getTimeUnit(),
|
||||
taskQueue,
|
||||
initParam.getThreadFactory(),
|
||||
initParam.rejectedExecutionHandler);
|
||||
taskQueue.setExecutor(fastThreadPoolExecutor);
|
||||
return fastThreadPoolExecutor;
|
||||
}
|
||||
|
||||
@Data
|
||||
@Accessors(chain = true)
|
||||
public static class ThreadPoolInitParam {
|
||||
|
||||
/**
|
||||
* 核心线程数量
|
||||
*/
|
||||
private Integer corePoolNum;
|
||||
|
||||
/**
|
||||
* 最大线程数量
|
||||
*/
|
||||
private Integer maxPoolNum;
|
||||
|
||||
/**
|
||||
* 线程存活时间
|
||||
*/
|
||||
private Long keepAliveTime;
|
||||
|
||||
/**
|
||||
* 线程存活时间单位
|
||||
*/
|
||||
private TimeUnit timeUnit;
|
||||
|
||||
/**
|
||||
* 队列最大容量
|
||||
*/
|
||||
private Integer capacity;
|
||||
|
||||
/**
|
||||
* 阻塞队列
|
||||
*/
|
||||
private BlockingQueue<Runnable> workQueue;
|
||||
|
||||
/**
|
||||
* 线程池任务满时拒绝任务策略
|
||||
*/
|
||||
private RejectedExecutionHandler rejectedExecutionHandler;
|
||||
|
||||
/**
|
||||
* 创建线程工厂
|
||||
*/
|
||||
private ThreadFactory threadFactory;
|
||||
|
||||
public ThreadPoolInitParam(String threadNamePrefix, boolean isDaemon) {
|
||||
this.threadFactory = new ThreadFactoryBuilder()
|
||||
.setNamePrefix(threadNamePrefix + "-")
|
||||
.setDaemon(isDaemon)
|
||||
.build();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,65 @@
|
||||
package io.dynamic.threadpool.starter.toolkit.thread;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.concurrent.RejectedExecutionException;
|
||||
import java.util.concurrent.RejectedExecutionHandler;
|
||||
import java.util.concurrent.ThreadFactory;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/**
|
||||
* 快速执行任务线程池, 参考 Dubbo 线程模型 EagerThreadPoolExecutor
|
||||
* <p>
|
||||
* 配合 {@link TaskQueue}
|
||||
*
|
||||
* @author chen.ma
|
||||
* @date 2021/7/5 21:00
|
||||
*/
|
||||
@Slf4j
|
||||
public class FastThreadPoolExecutor extends ThreadPoolExecutorTemplate {
|
||||
|
||||
public FastThreadPoolExecutor(int corePoolSize,
|
||||
int maximumPoolSize,
|
||||
long keepAliveTime,
|
||||
TimeUnit unit,
|
||||
TaskQueue<Runnable> workQueue,
|
||||
ThreadFactory threadFactory,
|
||||
RejectedExecutionHandler handler) {
|
||||
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
|
||||
}
|
||||
|
||||
private final AtomicInteger submittedTaskCount = new AtomicInteger(0);
|
||||
|
||||
public int getSubmittedTaskCount() {
|
||||
return submittedTaskCount.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void afterExecute(Runnable r, Throwable t) {
|
||||
submittedTaskCount.decrementAndGet();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(Runnable command) {
|
||||
submittedTaskCount.incrementAndGet();
|
||||
try {
|
||||
super.execute(command);
|
||||
} catch (RejectedExecutionException rx) {
|
||||
final TaskQueue queue = (TaskQueue) super.getQueue();
|
||||
try {
|
||||
if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
|
||||
submittedTaskCount.decrementAndGet();
|
||||
throw new RejectedExecutionException("队列容量已满.", rx);
|
||||
}
|
||||
} catch (InterruptedException x) {
|
||||
submittedTaskCount.decrementAndGet();
|
||||
throw new RejectedExecutionException(x);
|
||||
}
|
||||
} catch (Exception t) {
|
||||
submittedTaskCount.decrementAndGet();
|
||||
throw t;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,56 @@
|
||||
package io.dynamic.threadpool.starter.toolkit.thread;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.RejectedExecutionHandler;
|
||||
|
||||
/**
|
||||
* 线程池拒绝策略
|
||||
*
|
||||
* @author chen.ma
|
||||
* @date 2021/7/5 21:23
|
||||
*/
|
||||
@Slf4j
|
||||
public class RejectedPolicies {
|
||||
|
||||
/**
|
||||
* 发生拒绝事件时, 添加新任务并运行最早的任务
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static RejectedExecutionHandler runsOldestTaskPolicy() {
|
||||
return (r, executor) -> {
|
||||
if (executor.isShutdown()) {
|
||||
return;
|
||||
}
|
||||
BlockingQueue<Runnable> workQueue = executor.getQueue();
|
||||
Runnable firstWork = workQueue.poll();
|
||||
boolean newTaskAdd = workQueue.offer(r);
|
||||
if (firstWork != null) {
|
||||
firstWork.run();
|
||||
}
|
||||
if (!newTaskAdd) {
|
||||
executor.execute(r);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用阻塞方法将拒绝任务添加队列, 可保证任务不丢失
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
public static RejectedExecutionHandler syncPutQueuePolicy() {
|
||||
return (r, executor) -> {
|
||||
if (executor.isShutdown()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
executor.getQueue().put(r);
|
||||
} catch (InterruptedException e) {
|
||||
log.error("线程池添加队列任务失败", e);
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,52 @@
|
||||
package io.dynamic.threadpool.starter.toolkit.thread;
|
||||
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.RejectedExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* 快速执行任务而阻塞队列, 参考 Dubbo 重写队列 TaskQueue
|
||||
* <p>
|
||||
* 配合 {@link FastThreadPoolExecutor} 使用
|
||||
*
|
||||
* @author chen.ma
|
||||
* @date 2021/7/5 21:00
|
||||
*/
|
||||
public class TaskQueue<R extends Runnable> extends LinkedBlockingQueue<Runnable> {
|
||||
|
||||
private static final long serialVersionUID = -2635853580887179627L;
|
||||
|
||||
private FastThreadPoolExecutor executor;
|
||||
|
||||
public TaskQueue(int capacity) {
|
||||
super(capacity);
|
||||
}
|
||||
|
||||
public void setExecutor(FastThreadPoolExecutor exec) {
|
||||
executor = exec;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean offer(Runnable runnable) {
|
||||
int currentPoolThreadSize = executor.getPoolSize();
|
||||
// 如果有核心线程正在空闲, 将任务加入阻塞队列, 由核心线程进行处理任务
|
||||
if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
|
||||
return super.offer(runnable);
|
||||
}
|
||||
|
||||
// 当前线程池线程数量小于最大线程数, 返回false, 根据线程池源码, 会创建非核心线程
|
||||
if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 如果当前线程池数量大于最大线程数, 任务加入阻塞队列
|
||||
return super.offer(runnable);
|
||||
}
|
||||
|
||||
public boolean retryOffer(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
|
||||
if (executor.isShutdown()) {
|
||||
throw new RejectedExecutionException("执行器已关闭!");
|
||||
}
|
||||
return super.offer(o, timeout, unit);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,65 @@
|
||||
package io.dynamic.threadpool.starter.toolkit.thread;
|
||||
|
||||
import io.dynamic.threadpool.starter.toolkit.ArrayUtil;
|
||||
|
||||
import java.util.concurrent.*;
|
||||
|
||||
/**
|
||||
* 线程池创建模版
|
||||
*
|
||||
* @author chen.ma
|
||||
* @date 2021/7/5 21:59
|
||||
*/
|
||||
public class ThreadPoolExecutorTemplate extends ThreadPoolExecutor {
|
||||
|
||||
public ThreadPoolExecutorTemplate(int corePoolSize,
|
||||
int maximumPoolSize,
|
||||
long keepAliveTime,
|
||||
TimeUnit unit,
|
||||
BlockingQueue<Runnable> workQueue,
|
||||
ThreadFactory threadFactory,
|
||||
RejectedExecutionHandler handler) {
|
||||
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
|
||||
}
|
||||
|
||||
private Exception clientTrace() {
|
||||
return new Exception("tread task root stack trace");
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(final Runnable command) {
|
||||
super.execute(wrap(command, clientTrace()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Future<?> submit(final Runnable task) {
|
||||
return super.submit(wrap(task, clientTrace()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T> Future<T> submit(final Callable<T> task) {
|
||||
return super.submit(wrap(task, clientTrace()));
|
||||
}
|
||||
|
||||
private Runnable wrap(final Runnable task, final Exception clientStack) {
|
||||
return () -> {
|
||||
try {
|
||||
task.run();
|
||||
} catch (Exception e) {
|
||||
e.setStackTrace(ArrayUtil.addAll(clientStack.getStackTrace(), e.getStackTrace()));
|
||||
throw e;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private <T> Callable<T> wrap(final Callable<T> task, final Exception clientStack) {
|
||||
return () -> {
|
||||
try {
|
||||
return task.call();
|
||||
} catch (Exception e) {
|
||||
e.setStackTrace(ArrayUtil.addAll(clientStack.getStackTrace(), e.getStackTrace()));
|
||||
throw e;
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue