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

前言

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

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

public class PoolConfig {
 
 private int queueCapacity = 200;
 
 private int count = 0;
 
 private int maxCount = 0;
 
 private int aliveSec;
 
 public int getQueueCapacity() {
 return queueCapacity;
 } 
 
 public void setQueueCapacity(int queueCapacity) {
 this.queueCapacity = queueCapacity;
 }
 
 public void setCount(int count) {
 this.count = count;
 }
 
 public void setMaxCount(int maxCount) {
 this.maxCount = maxCount;
 }
 
 public void setAliveSec(int aliveSec) {
 this.aliveSec = aliveSec;
 }
 
 public int getCount() {
 return count;
 }
 
 public int getMaxCount() {
 return maxCount;
 }
 
 public int getAliveSec() {
 return aliveSec;
 }
}

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

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
 
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
 

@Component
@ConfigurationProperties(prefix="thread")
public class ThreadPoolConfig {
 
 private PoolConfig pool = new PoolConfig();
 
 Map count = new HashMap<>();
 
 public PoolConfig getPool() {
 return pool;
 }
 
 public void setPool(PoolConfig pool) {
 this.pool = pool;
 }
 
 public Map getCount() {
 return count;
 }
 
}

定义Task注解，方便使用：

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@documented
@Component
public @interface ExcutorTask {
 
 
 String value() default "";
 
}

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

public class Beans {
 
 private static final char PREFIX = '.';
 
 public static ConcurrentMap scanBeanClassNames(){
 ConcurrentMap beanClassNames = new ConcurrentHashMap<>();
 ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false); 
   provider.addIncludeFilter(new AnnotationTypeFilter(ExcutorTask.class));
   for(Package pkg : Package.getPackages()){
   String basePackage = pkg.getName();
     Set components = provider.findCandidateComponents(basePackage); 
     for (BeanDefinition component : components) {
     String beanClassName = component.getBeanClassName();
     try {
    Class clazz = Class.forName(component.getBeanClassName());
    boolean isAnnotationPresent = clazz.isAnnotationPresent(ZimaTask.class);
    if(isAnnotationPresent){
     ZimaTask task = clazz.getAnnotation(ExcutorTask.class);
     String aliasName = task.value();
     if(aliasName != null && !"".equals(aliasName)){
     beanClassNames.put(aliasName, component.getBeanClassName());
     }
    }
    } catch (ClassNotFoundException e) {
    e.printStackTrace();
    }
     beanClassNames.put(beanClassName.substring(beanClassName.lastIndexOf(PREFIX) + 1), component.getBeanClassName());
     }
   }
   return beanClassNames;
  } 
}

 线程执行类TaskPool：

@Component
public class TaskPool {
 
 public ThreadPoolTaskExecutor poolTaskExecutor;
 
 @Autowired 
 private ThreadPoolConfig threadPoolConfig;
 
 @Autowired 
 private ApplicationContext context;
 
 private final Integer MAX_POOL_SIZE = 2000;
 
 private PoolConfig poolCfg;
 
 private Map tasksCount;
 
 private ConcurrentMap beanClassNames;
 
 @PostConstruct
  public void init() {
 
 beanClassNames = Beans.scanBeanClassNames();
   
   poolTaskExecutor = new ThreadPoolTaskExecutor();
   
   poolCfg = threadPoolConfig.getPool();
 
 tasksCount = threadPoolConfig.getCount();
 
 int corePoolSize = poolCfg.getCount(), 
  maxPoolSize = poolCfg.getMaxCount(), 
  queueCapacity = poolCfg.getQueueCapacity(), 
  minPoolSize = 0, maxCount = (corePoolSize << 1);
 
 for(String taskName : 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 < corePoolSize){
  corePoolSize = MAX_POOL_SIZE;
  }
  poolTaskExecutor.setCorePoolSize(corePoolSize);
 }
 
 if(maxPoolSize > 0){
  if(maxPoolSize <= maxCount){
  maxPoolSize = maxCount;
  }
  if(MAX_POOL_SIZE < maxPoolSize){
  maxPoolSize = MAX_POOL_SIZE;
  }
  poolTaskExecutor.setMaxPoolSize(maxPoolSize);
 }
 
 if(poolCfg.getAliveSec() > 0){
  poolTaskExecutor.setKeepAliveSeconds(poolCfg.getAliveSec());
 }
 
 poolTaskExecutor.initialize();
  }
  
 public void execute(Class... clazz){
 int i = 0, len = tasksCount.size();
 for(; i < len; i++){
  Integer taskCount = tasksCount.get(i);
  for(int t = 0; t < taskCount; t++){
  try{
   Object taskObj = context.getBean(clazz[i]);
   if(taskObj != null){
   poolTaskExecutor.execute((Runnable) taskObj);
   }
  }catch(Exception ex){
   ex.printStackTrace();
  }
  }
 }
  }
  
 public void execute(String... args){
   int i = 0, len = tasksCount.size();
 for(; i < len; i++){
  Integer taskCount = tasksCount.get(i);
  for(int t = 0; t < taskCount; t++){
  try{
   Object taskObj = null;
   if(context.containsBean(args[i])){
   taskObj = context.getBean(args[i]);
   }else{
   if(beanClassNames.containsKey(args[i].toLowerCase())){
    Class clazz = Class.forName(beanClassNames.get(args[i].toLowerCase()));
    taskObj = context.getBean(clazz);
   }
   }
   if(taskObj != null){
   poolTaskExecutor.execute((Runnable) taskObj);
   }
  }catch(Exception ex){
   ex.printStackTrace();
  }
  }
 }
  }
 
 public void execute(){
 for(String taskName : tasksCount.keySet()){
  Integer taskCount = tasksCount.get(taskName);
  for(int t = 0; t < taskCount; t++){
  try{
   Object taskObj = null;
   if(context.containsBean(taskName)){
   taskObj = context.getBean(taskName);
   }else{
   if(beanClassNames.containsKey(taskName)){
    Class clazz = Class.forName(beanClassNames.get(taskName));
    taskObj = context.getBean(clazz);
   }
   }
   if(taskObj != null){
   poolTaskExecutor.execute((Runnable) taskObj);
   }
  }catch(Exception ex){
   ex.printStackTrace();
  }
  }
 }
  }
  
}

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

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

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

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

@Configuration
public class TaskManager {
 
 @Resource
 private TaskPool taskPool;
 
 @PostConstruct
 public void executor(){
 taskPool.execute();
 }
}

3.具体使用

@ExcutorTask
public class NeedExcutorTask implements Runnable{
  @Override
 public void run() {
    Thread.sleep(1000L);
    log.info("====== 任务执行 =====")
  }
}

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