java annotation example of how to implement a configurable thread pool
- 2021-06-28 12:24:21
- OfStack
Preface
The project requires multithreaded execution of 1 Task for ease of use of individual services.Intentionally encapsulated a public tool class, directly below the code:
PoolConfig (Thread Pool Core Configuration Parameter):
/**
* <h1> Thread pool core configuration (<b style="color:#CD0000"> Number of base thread pools, maximum thread pools, queue initial capacity, thread connection keep active seconds (default) 60s ) </b>)</h1>
*
* <blockquote><code>
* <table border="1px" style="border-color:gray;" width="100%"><tbody>
* <tr><th style="color:green;text-align:left;">
* Property Name
* </th><th style="color:green;text-align:left;">
* Attribute Meaning
* </th></tr>
* <tr><td>
* queueCapacity
* </td><td>
* Number of base thread pools
* </td></tr>
* <tr><td>
* count
* </td><td>
* Maximum number of thread pools
* </td></tr>
* <tr><td>
* maxCount
* </td><td>
* Queue Initial Capacity
* </td></tr>
* <tr><td>
* aliveSec
* </td><td>
* Thread connections remain active seconds (default) 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 (Thread pool configuration yml configuration item starts with 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> Thread pool configuration (<b style="color:#CD0000"> Thread pool core configuration, number of tasks processed by each business </b>)</h1>
*
* <blockquote><code>
* <table border="1px" style="border-color:gray;" width="100%"><tbody>
* <tr><th style="color:green;text-align:left;">
* Property Name
* </th><th style="color:green;text-align:left;">
* Attribute Meaning
* </th></tr>
* <tr><td>
* pool
* </td><td>
* Thread pool core configuration
* [ {@link PoolConfig} ]
* </td></tr>
* <tr><td>
* count
* </td><td>
* Initial number of tasks for each business task in the thread pool
* </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;
}
}Define the Task comment for ease of use:
@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 "";
}
Get a set of tasks using the Task annotation through reflection:
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;
}
}Thread execution class 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();
}
}
}
}
}
How to use it? (Everything needs a complete set of ^^)
1. Because you are using the springboot project, you need to add application.properties or application.yml
# Configuration Execution task Number of threads
thread.count.NeedExcutorTask=4
# Maximum survival time
thread.pool.aliveSec=300000
# Other configurations are the same
2. Load the thread configuration we wrote into our project
@Configuration
public class TaskManager {
@Resource
private TaskPool taskPool;
@PostConstruct
public void executor(){
taskPool.execute();
}
}
3. Use Specifically
@ExcutorTask
public class NeedExcutorTask implements Runnable{
@Override
public void run() {
Thread.sleep(1000L);
log.info("====== Task Execution =====")
}
}