java 注解实现一个可配置线程池的方法示例
前言
项目需要多线程执行一些Task,为了方便各个服务的使用。特意封装了一个公共工具类,下面直接撸代码:
PoolConfig(线程池核心配置参数):
/** *线程池核心配置(
* *基本线程池数量、最大线程池数量、队列初始容量、线程连接保持活动秒数(默认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; } }*
* * *属性名称 * *属性含义 * * *queueCapacity * *基本线程池数量 * * *count * *最大线程池数量 * * *maxCount * *队列初始容量 * * * *aliveSec * *线程连接保持活动秒数(默认60s) *
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; /** *线程池配置(
* *线程池核心配置、各个业务处理的任务数量) */ @Component @ConfigurationProperties(prefix="thread") publicclassThreadPoolConfig{ privatePoolConfigpool=newPoolConfig(); Map*
* * *属性名称 * *属性含义 * * *pool * *线程池核心配置 *【{@linkPoolConfig}】 * * * *count * *线程池各个业务任务初始的任务数 * count=newHashMap<>(); publicPoolConfiggetPool(){ returnpool; } publicvoidsetPool(PoolConfigpool){ this.pool=pool; } publicMap getCount(){ returncount; } }
定义Task注解,方便使用:
@Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Component public@interfaceExcutorTask{ /** *ThevaluemayindicateasuggestionforalogicalExcutorTaskname, *tobeturnedintoaSpringbeanincaseofanautodetectedExcutorTask. *@returnthesuggestedExcutorTaskname,ifany */ Stringvalue()default""; }
通过反射获取使用Task注解的任务集合:
publicclassBeans{ privatestaticfinalcharPREFIX='.'; publicstaticConcurrentMapscanBeanClassNames(){ ConcurrentMap beanClassNames=newConcurrentHashMap<>(); ClassPathScanningCandidateComponentProviderprovider=newClassPathScanningCandidateComponentProvider(false); provider.addIncludeFilter(newAnnotationTypeFilter(ExcutorTask.class)); for(Packagepkg:Package.getPackages()){ StringbasePackage=pkg.getName(); Set components=provider.findCandidateComponents(basePackage); for(BeanDefinitioncomponent:components){ StringbeanClassName=component.getBeanClassName(); try{ Class>clazz=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; privateConcurrentMap beanClassNames; @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_SIZE 0){ if(maxPoolSize<=maxCount){ maxPoolSize=maxCount; } if(MAX_POOL_SIZE 0){ poolTaskExecutor.setKeepAliveSeconds(poolCfg.getAliveSec()); } poolTaskExecutor.initialize(); } publicvoidexecute(Class>...clazz){ inti=0,len=tasksCount.size(); for(;i clazz=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;t clazz=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("======任务执行=====") } }
以上就是创建一个可扩展的线程池相关的配置(望指教~~~)。希望对大家的学习有所帮助,也希望大家多多支持毛票票。