How Spring Boot uses the thread pool to handle tens of thousands of data insertion functions

# Preface

When I was working on a project two days ago, I wanted to improve the performance optimization of inserting tables. Since there are two tables, the old table should be inserted first, followed by the new table, which costs more than 10,000 The data is a bit slow.

I thought of the thread pool ThreadPoolExecutor later, but using the Spring Boot project, you can use the thread pool ThreadPoolTaskExecutor provided by Spring to encapsulate the ThreadPoolExecutor, and directly use annotations to enable it

# Steps to use

First create a thread pool configuration and let Spring Boot load it to define how to create a ThreadPoolTaskExecutor. Use the two annotations @Configuration and @EnableAsync to indicate This is a configuration class, and it is the configuration class of the thread pool

@Configuration
@EnableAsync
public class ExecutorConfig {
    private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class);
    @Value("${async.executor.thread.core_pool_size}")    
    private int corePoolSize;   
    
    @Value("${async.executor.thread.max_pool_size}")    
    private int maxPoolSize;   
    
    @Value("${async.executor.thread.queue_capacity}")  
    private int queueCapacity;   
    
    @Value("${async.executor.thread.name.prefix}")  
    private String namePrefix;
    @Bean(name = "asyncServiceExecutor")    
    public Executor asyncServiceExecutor() {   
        logger.info("start asyncServiceExecutor");    
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); 
    
        //配置核心线程数       
        executor.setCorePoolSize(corePoolSize);   
    
        //配置最大线程数      
        executor.setMaxPoolSize(maxPoolSize);   
    
        //配置队列大小     
        executor.setQueueCapacity(queueCapacity);    
    
        //配置线程池中的线程的名称前缀        
        executor.setThreadNamePrefix(namePrefix);
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务        
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行  
         
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());     
        //执行初始化      
        executor.initialize();   
        return executor;  
     }
}

@Value is configured by me in application.properties. You can refer to the configuration and freely define

# 异步线程配置
# 配置核心线程数
async.executor.thread.core_pool_size = 5
# 配置最大线程数
async.executor.thread.max_pool_size = 5
# 配置队列大小
async.executor.thread.queue_capacity = 99999
# 配置线程池中的线程的名称前缀
async.executor.thread.name.prefix = async-service-

Create a Service interface, which is an asynchronous thread Interface

public interface AsyncService {   
    /**     
      * 执行异步任务     
      * 可以根据需求，自己加参数拟定，我这里就做个测试演示    
      */   
    void executeAsync();
}

Implementation class

@Service
public class AsyncServiceImpl implements AsyncService {  
    private static final Logger logger = LoggerFactory.getLogger(AsyncServiceImpl.class);
    @Override 
    @Async("asyncServiceExecutor")    
    public void executeAsync() {    
        logger.info("start executeAsync");
        System.out.println("异步线程要做的事情");        
        System.out.println("可以在这里执行批量插入等耗时的事情");
        logger.info("end executeAsync");   
    }
}

Add the annotation @Async("asyncServiceExecutor") to the executeAsync() method. The asyncServiceExecutor method is the method name in the previous ExecutorConfig.java, indicating that the executeAsync method enters The thread pool is created by the asyncServiceExecutor method

The next step is to inject the Service through the annotation @Autowired in the Controller or somewhere

@Autowiredprivate 
AsyncService asyncService;

@GetMapping("/async")
public void async(){  
    asyncService.executeAsync();
}

Log printing

2022- 07-16 22:15:47.655 INFO 10516 --- [async-service-5] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
Things to be done by asynchronous threads
You can perform batch insertion and other time-consuming tasks here Things
2022-07-16 22:15:47.655 INFO 10516 --- [async-service-5] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync
2022-07-16 22:15:47.770 INFO 10516 --- [async-service-1] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
Things to be done by asynchronous threads
Time-consuming things such as batch insertion can be performed here
2022-07- 16 22:15:47.770 INFO 10516 --- [async-service-1] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync
2022-07-16 22:15:47.816 INFO 10516 --- [async-service- 2] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
What the asynchronous thread has to do
You can perform time-consuming things such as batch insertion here
2022-07-16 22:15:47.816 INFO 10516 - -- [async-service-2] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync
2022-07-16 22:15:48.833 INFO 10516 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
Things to be done by asynchronous threads
You can perform time-consuming tasks such as batch insertion here
2022-07-16 22:15:48.834 INFO 10516 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync
2022-07-16 22:15:48.986 INFO 10516 --- [async-service-4] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
Asynchronous thread to do Things
You can perform time-consuming things such as batch insertion here
2022-07-16 22:15:48.987 INFO 10516 --- [async-service-4] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync

It can be found from the above log that [async-service-] has multiple threads, which are obviously executed in the thread pool we configured, and in each request, the controller starts and end logs are printed continuously, indicating that each request is responded to quickly, and time-consuming operations are left to the threads in the thread pool for asynchronous execution;

Although we have used the thread pool, It is still unclear what the situation of the thread pool was at that time. How many threads were executing and how many were waiting in the queue? Here I created a subclass of ThreadPoolTaskExecutor, which will print out the running status of the current thread pool every time a thread is submitted

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;
import java.util.concurrent.Callable;import java.util.concurrent.Future;import java.util.concurrent.ThreadPoolExecutor;
/** 
* @Author: 腾腾 
* @Date: 2022/7/16/0016 22:19 
*/
public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
    private static final Logger logger = LoggerFactory.getLogger(VisiableThreadPoolTaskExecutor.class);
    private void showThreadPoolInfo(String prefix) {        
        ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();
        if (null == threadPoolExecutor) {    
            return;  
        }
        logger.info("{}, {},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]",                
        this.getThreadNamePrefix(),         
        prefix,           
        threadPoolExecutor.getTaskCount(),     
        threadPoolExecutor.getCompletedTaskCount(),  
        threadPoolExecutor.getActiveCount(),    
        threadPoolExecutor.getQueue().size());
     }
    @Override    
    public void execute(Runnable task) {    
        showThreadPoolInfo("1. do execute");       
        super.execute(task);   
    }
    @Override    
    public void execute(Runnable task, long startTimeout) {      
        showThreadPoolInfo("2. do execute");   
        super.execute(task, startTimeout);  
    }
    @Override  
    public Future<?> submit(Runnable task) {   
        showThreadPoolInfo("1. do submit");    
        return super.submit(task);   
    }
    @Override  
    public <T> Future<T> submit(Callable<T> task) {   
        showThreadPoolInfo("2. do submit");      
        return super.submit(task); 
    }
    @Override    
    public ListenableFuture<?> submitListenable(Runnable task) {    
        showThreadPoolInfo("1. do submitListenable");   
        return super.submitListenable(task);   
    }
    @Override
    public <T> ListenableFuture<T> submitListenable(Callable<T> task) {     
        showThreadPoolInfo("2. do submitListenable");     
        return super.submitListenable(task);  
    }
}

As shown above, the showThreadPoolInfo method will print out the total number of tasks, the number of completed tasks, the The number of threads and queue size are printed out, and then the execute, submit and other methods of the parent class are overridden, and the showThreadPoolInfo method is called inside, so that every time a task is submitted to the thread pool, the basic situation of the current thread pool will be printed. to the log.

Modify the asyncServiceExecutor method of ExecutorConfig.java and change ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor() to ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor()

@Bean(name = "asyncServiceExecutor")    
public Executor asyncServiceExecutor() {  
    logger.info("start asyncServiceExecutor");  
    //在这里修改       
    ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();    
    //配置核心线程数     
    executor.setCorePoolSize(corePoolSize);  
    //配置最大线程数      
    executor.setMaxPoolSize(maxPoolSize);  
    //配置队列大小      
    executor.setQueueCapacity(queueCapacity); 
    //配置线程池中的线程的名称前缀      
    executor.setThreadNamePrefix(namePrefix);
    
    // rejection-policy：当pool已经达到max size的时候，如何处理新任务      
    // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行     
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); 
    //执行初始化       
    executor.initialize();       
    return executor; 
}

Start the project test again

2022-07-16 22:23:30.951 INFO 14088 --- [nio-8087-exec-2] u.d.e.e.i.VisiableThreadPoolTaskExecutor: async-service-, 2. do submit,taskCount [0], completedTaskCount [ 0], activeCount [0], queueSize [0]
2022-07-16 22:23:30.952 INFO 14088 --- [async-service-1] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
asynchronous What the thread has to do
You can perform time-consuming tasks such as batch insertion here
2022-07-16 22:23:30.953 INFO 14088 --- [async-service-1] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync
2022-07-16 22:23:31.351 INFO 14088 --- [nio-8087-exec-3] u.d.e.e.i.VisiableThreadPoolTaskExecutor : async-service-, 2. do submit,taskCount [1], completedTaskCount [ 1], activeCount [0], queueSize [0]
2022-07-16 22:23:31.353 INFO 14088 --- [async-service-2] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
asynchronous What the thread has to do
You can perform time-consuming tasks such as batch insertion here
2022-07-16 22:23:31.353 INFO 14088 --- [async-service-2] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync
2022-07-16 22:23:31.927 INFO 14088 --- [nio-8087-exec-5] u.d.e.e.i.VisiableThreadPoolTaskExecutor : async-service-, 2. do submit,taskCount [2], completedTaskCount [ 2], activeCount [0], queueSize [0]
2022-07-16 22:23:31.929 INFO 14088 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
asynchronous What the thread has to do
You can perform time-consuming tasks such as batch insertion here
2022-07-16 22:23:31.930 INFO 14088 --- [async-service-3] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync
2022-07-16 22:23:32.496 INFO 14088 --- [nio-8087-exec-7] u.d.e.e.i.VisiableThreadPoolTaskExecutor : async-service-, 2. do submit,taskCount [3], completedTaskCount [ 3], activeCount [0], queueSize [0]
2022-07-16 22:23:32.498 INFO 14088 --- [async-service-4] c.u.d.e.executor.impl.AsyncServiceImpl : start executeAsync
asynchronous What the thread has to do
You can perform time-consuming tasks such as batch insertion here
2022-07-16 22:23:32.499 INFO 14088 --- [async-service-4] c.u.d.e.executor.impl.AsyncServiceImpl : end executeAsync

Note this line of log:

2022-07-16 22:23:32.496 INFO 14088 --- [nio-8087-exec-7] u.d.e.e.i.VisiableThreadPoolTaskExecutor: async-service-, 2. do submit,taskCount [3], completedTaskCount [3], activeCount [0], queueSize [0]

This shows that when submitting a task to the thread pool , the method submit(Callable task) is called. Currently, 3 tasks have been submitted and 3 have been completed. There are currently 0 threads processing tasks and 0 tasks left waiting in the queue. The basic situation of the thread pool is Clearly.

The above is the detailed content of How Spring Boot uses the thread pool to handle tens of thousands of data insertion functions. For more information, please follow other related articles on the PHP Chinese website!