前言

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

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

/** * <h1>线程池核心配置(<b style="color:#CD0000">基本线程池数量、最大线程池数量、队列初始容量、线程连接保持活动秒数（默认60s）</b>)</h1> * * <blockquote><code> * <table border="1px" style="border-color:gray;" width="100%"><tbody> * <tr><th style="color:green;text-align:left;"> * 属性名称 * </th><th style="color:green;text-align:left;"> * 属性含义 * </th></tr> * <tr><td> * queueCapacity * </td><td> * 基本线程池数量 * </td></tr> * <tr><td> * count * </td><td> * 最大线程池数量 * </td></tr> * <tr><td> * maxCount * </td><td> * 队列初始容量 * </td></tr> * <tr><td> * aliveSec * </td><td> * 线程连接保持活动秒数（默认60s） * </td></tr> * </tbody></table> * </code></blockquote> */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; /** * <h1>线程池配置(<b style="color:#CD0000">线程池核心配置、各个业务处理的任务数量</b>)</h1> * * <blockquote><code> * <table border="1px" style="border-color:gray;" width="100%"><tbody> * <tr><th style="color:green;text-align:left;"> * 属性名称 * </th><th style="color:green;text-align:left;"> * 属性含义 * </th></tr> * <tr><td> * pool * </td><td> * 线程池核心配置 * 【{@link PoolConfig}】 * </td></tr> * <tr><td> * count * </td><td> * 线程池各个业务任务初始的任务数 * </td></tr> * </tbody></table> * </code></blockquote> */@Component@ConfigurationProperties(prefix="thread")public class ThreadPoolConfig { private PoolConfig pool = new PoolConfig(); Map<String, Integer> count = new HashMap<>(); public PoolConfig getPool() { return pool; } public void setPool(PoolConfig pool) { this.pool = pool; } public Map<String, Integer> getCount() { return count; } }

定义Task注解，方便使用：

@Target(ElementType.TYPE)@Retention(RetentionPolicy.RUNTIME)@Documented@Componentpublic @interface ExcutorTask { /** * The value may indicate a suggestion for a logical ExcutorTask name, * to be turned into a Spring bean in case of an autodetected ExcutorTask . * @return the suggested ExcutorTask name, if any */ String value() default ""; }

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

public class Beans { private static final char PREFIX = '.'; public static ConcurrentMap<String, String> scanBeanClassNames(){ ConcurrentMap<String, String> beanClassNames = new ConcurrentHashMap<>(); ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false); provider.addIncludeFilter(new AnnotationTypeFilter(ExcutorTask.class)); for(Package pkg : Package.getPackages()){ String basePackage = pkg.getName(); Set<BeanDefinition> 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：

@Componentpublic 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<String, Integer> tasksCount; private ConcurrentMap<String, String> 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.将我们写的线程配置进行装载到我们的项目中

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

3.具体使用

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

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