package com.yami.trading.common.config;
|
|
import java.lang.reflect.Constructor;
|
import java.util.Map;
|
import java.util.concurrent.Executor;
|
import java.util.concurrent.RejectedExecutionHandler;
|
import java.util.concurrent.ThreadPoolExecutor;
|
|
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
|
import org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler;
|
import org.springframework.beans.factory.BeanFactory;
|
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
|
import com.yami.trading.common.util.ApplicationUtil;
|
|
/**
|
* @author JORGE
|
* @description 线程池
|
*/
|
@SuppressWarnings({"unchecked","rawtypes"})
|
public class ThreadPool {
|
/**
|
* 默认应用线程池
|
*/
|
private static ThreadPoolTaskExecutor defaultThreadPool;
|
|
/**
|
* 允许核心线程超时后也关闭
|
*/
|
private static final Boolean ALLOW_CORETHREAD_TIMEOUT=false;
|
|
/**
|
* 新创建线程的名称前缀
|
*/
|
private static final String DEFAULT_THREAD_NAME_PREFIX="Application-ThreadPool-";
|
|
/**
|
* 丢弃新任务且抛出RejectedExecutionException异常(默认值)
|
*/
|
public static final Class<? extends RejectedExecutionHandler> DROP_THROWS=ThreadPoolExecutor.AbortPolicy.class;
|
|
/**
|
* 丢弃新任务且不抛出异常
|
*/
|
public static final Class<? extends RejectedExecutionHandler> DROP_NOTHROWS=ThreadPoolExecutor.DiscardPolicy.class;
|
|
/**
|
* 使用调用者线程处理队列溢出任务
|
*/
|
public static final Class<? extends RejectedExecutionHandler> CALLER_RUNS=ThreadPoolExecutor.CallerRunsPolicy.class;
|
|
/**
|
* 丢弃队列中老任务(最早,最前面)且不抛出异常
|
*/
|
public static final Class<? extends RejectedExecutionHandler> DROP_OLDS_NOTHROWS=ThreadPoolExecutor.DiscardOldestPolicy.class;
|
|
/**
|
* 获取Spring上下文中第一个兼容的线程池
|
* @param threadPoolType 兼容的线程池类型
|
* @return 线程池
|
*/
|
public static <R extends Executor> R getSpringThreadPool(Class<R>... threadPoolType){
|
Class<R> poolType=(null==threadPoolType||0==threadPoolType.length)?
|
(Class<R>)ThreadPoolTaskExecutor.class:threadPoolType[0];
|
Map<String, R> executors=ApplicationUtil.getBeansOfType(poolType);
|
if(null==executors||0==executors.size()) return null;
|
return executors.entrySet().iterator().next().getValue();
|
}
|
|
/**
|
* 获取应用线程池
|
* @return 应用线程池
|
*/
|
public static final <R extends Executor> R getApplicationThreadPool(){
|
if(null!=defaultThreadPool) return (R)defaultThreadPool;
|
|
int maxRevise=ApplicationUtil.getProperty("thread.pool.max.revise",Integer.class,0);
|
double maxFactor=ApplicationUtil.getProperty("thread.pool.max.factor",Double.class,2.0);
|
int queueCapacity=ApplicationUtil.getProperty("thread.pool.queue.size",Integer.class,200);
|
int keepAliveSeconds=ApplicationUtil.getProperty("thread.pool.max.idle",Integer.class,180);
|
|
int corePoolSize=Runtime.getRuntime().availableProcessors();
|
int maxPoolSize=(int)(maxFactor * corePoolSize + maxRevise);
|
return (R)(defaultThreadPool=getTaskExecutor(corePoolSize,maxPoolSize,queueCapacity,keepAliveSeconds,DEFAULT_THREAD_NAME_PREFIX,ALLOW_CORETHREAD_TIMEOUT,CALLER_RUNS));
|
}
|
|
/**
|
* 获取异步调用异常处理器
|
* @return AsyncUncaughtExceptionHandler
|
*/
|
public static AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
|
return new SimpleAsyncUncaughtExceptionHandler();
|
}
|
|
/**
|
* 获取Sleuth中的LazyTraceExecutor包装器以支持自定义线程池中创建异步Span
|
* Sleuth默认的线程池包装器是org.springframework.cloud.sleuth.instrument.async.LazyTraceThreadPoolTaskExecutor
|
* @param executor 线程池执行器
|
* @return 包装后的线程池执行器
|
*/
|
public static Executor getLazyTraceExecutor(Executor executor){
|
try {
|
Class executorType=Class.forName("org.springframework.cloud.sleuth.instrument.async.LazyTraceExecutor");
|
Constructor<Executor> cons=executorType.getDeclaredConstructor(BeanFactory.class,Executor.class);
|
return cons.newInstance(ApplicationUtil.getApplicationContext(),executor);
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
return null;
|
}
|
|
/**
|
* 获取Sleuth中的LazyTraceThreadPoolTaskExecutor包装器以支持自定义线程池中创建异步Span
|
* Sleuth默认的线程池包装器是org.springframework.cloud.sleuth.instrument.async.LazyTraceThreadPoolTaskExecutor
|
* @param executor 线程池执行器
|
* @return 包装后的线程池执行器
|
*/
|
public static ThreadPoolTaskExecutor getLazyTraceExecutor(ThreadPoolTaskExecutor executor){
|
try {
|
Class executorType=Class.forName("org.springframework.cloud.sleuth.instrument.async.LazyTraceThreadPoolTaskExecutor");
|
Constructor<ThreadPoolTaskExecutor> cons=executorType.getDeclaredConstructor(BeanFactory.class,ThreadPoolTaskExecutor.class);
|
return cons.newInstance(ApplicationUtil.getApplicationContext(),executor);
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
return null;
|
}
|
|
/**
|
* 获取固定容量大小且无缓冲队列的线程池,池中无空闲线程后使用调用方线程处理任务
|
* @param poolSize 池中固定线程数量
|
* @param threadNamePrefixs 线程名称前缀
|
* @return
|
*/
|
public static ThreadPoolTaskExecutor getFixedTaskExecutor(Integer poolSize,String... threadNamePrefixs){
|
String threadNamePrefix=(null==threadNamePrefixs||0==threadNamePrefixs.length)?DEFAULT_THREAD_NAME_PREFIX:threadNamePrefixs[0];
|
if(null==threadNamePrefixs||0==threadNamePrefixs.length) {
|
threadNamePrefix=DEFAULT_THREAD_NAME_PREFIX;
|
}else {
|
threadNamePrefix=threadNamePrefixs[0];
|
if(null==threadNamePrefix||(threadNamePrefix=threadNamePrefix.trim()).isEmpty()) {
|
threadNamePrefix=DEFAULT_THREAD_NAME_PREFIX;
|
}
|
}
|
return getTaskExecutor(poolSize,poolSize,0,0,threadNamePrefix,ALLOW_CORETHREAD_TIMEOUT,CALLER_RUNS);
|
}
|
|
/**
|
* 获取固定容量大小且具有缓冲队列的线程池,池中无空闲线程后转入缓冲队列,队列溢出后使用调用方线程处理任务
|
* @param poolSize 池中固定线程数量
|
* @param queueSize 任务队列尺寸
|
* @param threadNamePrefixs 线程名称前缀
|
* @return 线程池
|
*/
|
public static ThreadPoolTaskExecutor getFixedTaskExecutor(Integer poolSize,Integer queueSize,String... threadNamePrefixs){
|
String threadNamePrefix=(null==threadNamePrefixs||0==threadNamePrefixs.length)?DEFAULT_THREAD_NAME_PREFIX:threadNamePrefixs[0];
|
if(null==threadNamePrefixs||0==threadNamePrefixs.length) {
|
threadNamePrefix=DEFAULT_THREAD_NAME_PREFIX;
|
}else {
|
threadNamePrefix=threadNamePrefixs[0];
|
if(null==threadNamePrefix||(threadNamePrefix=threadNamePrefix.trim()).isEmpty()) {
|
threadNamePrefix=DEFAULT_THREAD_NAME_PREFIX;
|
}
|
}
|
return getTaskExecutor(poolSize,poolSize,queueSize,0,threadNamePrefix,ALLOW_CORETHREAD_TIMEOUT,CALLER_RUNS);
|
}
|
|
/**
|
* 获取自定义线程池
|
* @param coreSize 池中核心线程数量
|
* @param maxSize 池中最大线程数量
|
* @param queueSize 任务队列尺寸
|
* @param idleTime 线程空闲时间
|
* @param threadNamePrefix 创建新线程的名称前缀
|
* @param allowCoreTimeout 允许核心数量的线程超时关闭
|
* @param rejectedPolicy 任务队列溢出处理策略
|
* @return
|
*/
|
public static ThreadPoolTaskExecutor getTaskExecutor(Integer coreSize,Integer maxSize,Integer queueSize,Integer idleTime,String threadNamePrefix,Boolean allowCoreTimeout,Class<? extends RejectedExecutionHandler> rejectedPolicy){
|
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
executor.setCorePoolSize(coreSize);
|
executor.setMaxPoolSize(maxSize);
|
executor.setQueueCapacity(queueSize);
|
executor.setKeepAliveSeconds(idleTime);
|
executor.setThreadNamePrefix(threadNamePrefix);
|
executor.setAllowCoreThreadTimeOut(allowCoreTimeout);
|
|
try {
|
executor.setRejectedExecutionHandler(rejectedPolicy.newInstance());
|
} catch (Exception e) {
|
throw new RuntimeException(e);
|
}
|
|
executor.afterPropertiesSet();
|
Runtime.getRuntime().addShutdownHook(new Thread(()->executor.shutdown()));
|
|
return executor;
|
}
|
}
|
