java 注解实现一个可配置线程池的方法示例

2023-08-29 11:08:04 311

前言

项目需要多线程执行一些Task，为了方便各个服务的使用。特意封装了一个公共工具类，下面直接撸代码：

PoolConfig(线程池核心配置参数):

/**
*线程池核心配置(基本线程池数量、最大线程池数量、队列初始容量、线程连接保持活动秒数（默认60s）)
*
*
*
*
*属性名称
*
*属性含义
*
*
*queueCapacity
*
*基本线程池数量
*
*
*count
*
*最大线程池数量
*
*
*maxCount
*
*队列初始容量
*
*
*aliveSec
*
*线程连接保持活动秒数（默认60s）
*
*
*

*/
publicclassPoolConfig{

privateintqueueCapacity=200;

privateintcount=0;

privateintmaxCount=0;

privateintaliveSec;

publicintgetQueueCapacity(){
returnqueueCapacity;
}

publicvoidsetQueueCapacity(intqueueCapacity){
this.queueCapacity=queueCapacity;
}

publicvoidsetCount(intcount){
this.count=count;
}

publicvoidsetMaxCount(intmaxCount){
this.maxCount=maxCount;
}

publicvoidsetAliveSec(intaliveSec){
this.aliveSec=aliveSec;
}

publicintgetCount(){
returncount;
}

publicintgetMaxCount(){
returnmaxCount;
}

publicintgetAliveSec(){
returnaliveSec;
}
}

ThreadPoolConfig(线程池配置yml配置项以thread开头):

importjava.util.ArrayList;
importjava.util.HashMap;
importjava.util.List;
importjava.util.Map;

importorg.springframework.boot.context.properties.ConfigurationProperties;
importorg.springframework.stereotype.Component;

/**
*线程池配置(线程池核心配置、各个业务处理的任务数量)
*
*
*
*
*属性名称
*
*属性含义
*
*
*pool
*
*线程池核心配置
*【{@linkPoolConfig}】
*
*
*count
*
*线程池各个业务任务初始的任务数
*
*
*

*/
@Component
@ConfigurationProperties(prefix="thread")
publicclassThreadPoolConfig{

privatePoolConfigpool=newPoolConfig();

Mapcount=newHashMap<>();

publicPoolConfiggetPool(){
returnpool;
}

publicvoidsetPool(PoolConfigpool){
this.pool=pool;
}

publicMapgetCount(){
returncount;
}

}

定义Task注解，方便使用：

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Component
public@interfaceExcutorTask{

/**
*ThevaluemayindicateasuggestionforalogicalExcutorTaskname,
*tobeturnedintoaSpringbeanincaseofanautodetectedExcutorTask.
*@returnthesuggestedExcutorTaskname,ifany
*/
Stringvalue()default"";

}

通过反射获取使用Task注解的任务集合：

publicclassBeans{

privatestaticfinalcharPREFIX='.';

publicstaticConcurrentMapscanBeanClassNames(){
ConcurrentMapbeanClassNames=newConcurrentHashMap<>();
ClassPathScanningCandidateComponentProviderprovider=newClassPathScanningCandidateComponentProvider(false);
provider.addIncludeFilter(newAnnotationTypeFilter(ExcutorTask.class));
for(Packagepkg:Package.getPackages()){
StringbasePackage=pkg.getName();
Setcomponents=provider.findCandidateComponents(basePackage);
for(BeanDefinitioncomponent:components){
StringbeanClassName=component.getBeanClassName();
try{
Classclazz=Class.forName(component.getBeanClassName());
booleanisAnnotationPresent=clazz.isAnnotationPresent(ZimaTask.class);
if(isAnnotationPresent){
ZimaTasktask=clazz.getAnnotation(ExcutorTask.class);
StringaliasName=task.value();
if(aliasName!=null&&!"".equals(aliasName)){
beanClassNames.put(aliasName,component.getBeanClassName());
}
}
}catch(ClassNotFoundExceptione){
e.printStackTrace();
}
beanClassNames.put(beanClassName.substring(beanClassName.lastIndexOf(PREFIX)+1),component.getBeanClassName());
}
}
returnbeanClassNames;
}
}

线程执行类TaskPool：

@Component
publicclassTaskPool{

publicThreadPoolTaskExecutorpoolTaskExecutor;

@Autowired
privateThreadPoolConfigthreadPoolConfig;

@Autowired
privateApplicationContextcontext;

privatefinalIntegerMAX_POOL_SIZE=2000;

privatePoolConfigpoolCfg;

privateMaptasksCount;

privateConcurrentMapbeanClassNames;

@PostConstruct
publicvoidinit(){

beanClassNames=Beans.scanBeanClassNames();

poolTaskExecutor=newThreadPoolTaskExecutor();

poolCfg=threadPoolConfig.getPool();

tasksCount=threadPoolConfig.getCount();

intcorePoolSize=poolCfg.getCount(),
maxPoolSize=poolCfg.getMaxCount(),
queueCapacity=poolCfg.getQueueCapacity(),
minPoolSize=0,maxCount=(corePoolSize<<1);

for(StringtaskName:tasksCount.keySet()){
minPoolSize+=tasksCount.get(taskName);
}

if(corePoolSize>0){
if(corePoolSize<=minPoolSize){
corePoolSize=minPoolSize;
}
}else{
corePoolSize=minPoolSize;
}

if(queueCapacity>0){
poolTaskExecutor.setQueueCapacity(queueCapacity);
}

if(corePoolSize>0){
if(MAX_POOL_SIZE0){
if(maxPoolSize<=maxCount){
maxPoolSize=maxCount;
}
if(MAX_POOL_SIZE0){
poolTaskExecutor.setKeepAliveSeconds(poolCfg.getAliveSec());
}

poolTaskExecutor.initialize();
}

publicvoidexecute(Class...clazz){
inti=0,len=tasksCount.size();
for(;iclazz=Class.forName(beanClassNames.get(args[i].toLowerCase()));
taskObj=context.getBean(clazz);
}
}
if(taskObj!=null){
poolTaskExecutor.execute((Runnable)taskObj);
}
}catch(Exceptionex){
ex.printStackTrace();
}
}
}
}

publicvoidexecute(){
for(StringtaskName:tasksCount.keySet()){
IntegertaskCount=tasksCount.get(taskName);
for(intt=0;tclazz=Class.forName(beanClassNames.get(taskName));
taskObj=context.getBean(clazz);
}
}
if(taskObj!=null){
poolTaskExecutor.execute((Runnable)taskObj);
}
}catch(Exceptionex){
ex.printStackTrace();
}
}
}
}

}

如何使用？（做事就要做全套^_^）

1.因为使用的springboot项目，需要在application.properties或者application.yml添加

#配置执行的task线程数
thread.count.NeedExcutorTask=4
#最大存活时间
thread.pool.aliveSec=300000
#其他配置同理

2.将我们写的线程配置进行装载到我们的项目中

@Configuration
publicclassTaskManager{

@Resource
privateTaskPooltaskPool;

@PostConstruct
publicvoidexecutor(){
taskPool.execute();
}
}

3.具体使用

@ExcutorTask
publicclassNeedExcutorTaskimplementsRunnable{
@Override
publicvoidrun(){
Thread.sleep(1000L);
log.info("======任务执行=====")
}
}

以上就是创建一个可扩展的线程池相关的配置（望指教~~~）。希望对大家的学习有所帮助，也希望大家多多支持毛票票。

java 注解实现一个可配置线程池的方法示例

线程池核心配置(基本线程池数量、最大线程池数量、队列初始容量、线程连接保持活动秒数（默认60s）)

线程池配置(线程池核心配置、各个业务处理的任务数量)

热门推荐

随机推荐