java 注解实现一个可配置线程池

亦凉 2022-03-20 04:59 236阅读 0赞

**前言**

*项目需要多线程执行一些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
    @Component
    public @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**：

@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<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.将我们写的线程配置进行装载到我们的项目中

@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("====== 任务执行 =====")
        }
    }

***以上就是创建一个可扩展的线程池相关的配置（望指教~~~）。***

本文原文地址：[https://blog.csdn.net/u011663149/article/details/86497456][https_blog.csdn.net_u011663149_article_details_86497456]

[https_blog.csdn.net_u011663149_article_details_86497456]: https://blog.csdn.net/u011663149/article/details/86497456