Spring Boot怎麼使用執行緒池處理上萬個資料插入功能

# 前言

前兩天做專案的時候，想提高一下插入表的效能優化，因為是兩張表，先插舊的表，緊接著插新的表，一萬多條資料就有點慢了

後面就想到了線程池ThreadPoolExecutor，而用的是Spring Boot項目，可以用Spring提供的對ThreadPoolExecutor封裝的線程池ThreadPoolTask​​Executor，直接使用註解啟用

# 使用步驟

先建立一個執行緒池的配置，讓Spring Boot加載，用來定義如何建立一個ThreadPoolTask​​Executor，要使用@Configuration和@EnableAsync這兩個註解，表示這是個配置類，並且是線程池的配置類

@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是我配置在application.properties，可以參考配置，自由定義

# 异步线程配置
# 配置核心线程数
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-

創建一個Service接口，是異步線程的介面

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

實作類別

@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");   
    }
}

在executeAsync()方法上增加註解@Async("asyncServiceExecutor")，asyncServiceExecutor方法是前面ExecutorConfig.java中的方法名，表示executeAsync方法進入的執行緒池是asyncServiceExecutor方法所建立的

接下來就是在Controller裡或是哪裡透過註解@Autowired注入這個Service

@Autowiredprivate 
AsyncService asyncService;

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

日誌列印

# 2022- 07-16 22:15:47.655  INFO 10516 --- [async-service-5] c.u.d.e.executor.impl.AsyncServiceImpl   : start executeAsync
 AsyncServiceImpl   : start executeAsync
 AsyncServiceImpl   : start executeAsync
 .AsyncServiceImpl   : start executeAsync
 .AsyncServiceImpl   : start executeAsync
 .的事
 2022-07-16 22:15:47.655  INFO 10516 --- [async-service-5] c.u.d.e.executor.impl.AsyncServiceImpl  : end executeAsy2#75075075075:00712121F7:21075:0075-F 10516 --- [async-service-1] c.u.d.e.executor.impl.AsyncServiceImpl   : start executeAsync
 異步線程要做的事情
 可以在這裡執行批量插入等耗時的事情## 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
 非同步執行緒要做的事情
 可以在這裡執行批次插入等耗時的事情
 2022-07-16 22:15:47.8 -- [async-service-2] c.u.d.e.executor.impl.AsyncServiceImpl   : end executeAsync
 2022-07-16 22:15:48.833  INFO 10516 ---async. start executeAsync
 非同步執行緒要做的事情
 可以在這裡執行批次插入等耗時的事情
 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-48.986  INFO 10516 --- [async-service-4] c.u.d. # 異步線程要做的事情
 可以在這裡執行批次插入等耗時的事情
 2022-07-16 22:15:48.987  INFO 10516 --- [async-service-4] c.u.d.e.executor.impl.AsyncuteImimpl.Asyncute pl

透過以上日誌可以發現，[async-service-]是有多個線程的，顯然已經在我們配置的線程池中執行了，並且每次請求中，controller的起始和結束日誌都是連續列印的，表示每次請求都快速回應了，而耗時的操作都留給線程池中的線程去非同步執行；

雖然我們已經用上了線程池，但是目前還不清楚線程池當時的情況，有多少線程在執行，多少在隊列中等待？這裡我創建了一個ThreadPoolTask​​Executor的子類，在每次提交線程的時候都會將當前線程池的運行狀況打印出來

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);  
    }
}

如上所示，showThreadPoolInfo方法中將任務總數、已完成數、活躍執行緒數，佇列大小都印出來了，然後Override了父類別的execute、submit等方法，在裡面呼叫showThreadPoolInfo方法，這樣每次有任務被提交到執行緒池的時候，都會將目前執行緒池的基本情況列印到日誌中。

修改ExecutorConfig.java的asyncServiceExecutor方法，將ThreadPoolTask​​Executor executor = new ThreadPoolTask​​Executor()改為ThreadPoolTask​​Executor executor = new VisiableThreadPoolTask​​#ThreadPoolTask​​##re(askExecutor executor = new VisiableThreadPoolTask​​#ThreadPoolTask​​ #

2022-07-16 22:23:30.951  INFO 14088 --- [nio-8087-exec-2] u.d.e.e.i.VisiableThreadPoolTask​​Executor : asynd.e.e.i.VisiableThreadPoolTask​​Executor : async-do-, 2. 0], activeCount [0], queueSize [0]
2022-07-16 22:23:30.952  INFO 14088 --- [async-service-1] c.u.d.e.executor.impl.AsyService:Im執行緒要做的事情
可以在這裡執行批次插入等耗時的事情
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.VisiableThreadPoolTask​​Task7-exec-3] u.d.e.e.i。 1], activeCount [0], queueSize [0]
2022-07-16 22:23:31.353  INFO 14088 --- [async-service-2] c.u.d.e.executor.impl.AsyService:Im  執行緒要做的事情
可以在這裡執行批次插入等耗時的事情
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.VisiableThreadPoolTask​​Task7-exec-5] u.d.e.e.i。 2], activeCount [0], queueSize [0]
2022-07-16 22:23:31.929  INFO 14088 --- [async-service-3] c.u.d.e.executor.impl.AsyService:Im執行緒要做的事情
可以在這裡執行批次插入等耗時的事情
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.VisiableThreadPoolTask​​Task7-exec-7] u.d.e.e.i。 3], activeCount [0], queueSize [0]
2022-07-16 22:23:32.498  INFO 14088 --- [async-service-4] c.u.d.e.executor.impl.AsyService:Im  執行緒要做的事情
可以在這裡執行批次插入等耗時的事情
2022-07-16 22:23:32.499  INFO 14088 --- [async-service-4] 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.VisiableThreadPoolTask​​Executor : async-service-, 2. do submit,taskCount [3], completedTaskCount [3], activeCount [0], queueSize [0]

這說明提交任務到線程池的時候，呼叫的是submit(Callable task)這個方法，目前已經提交了3個任務，完成了3個，目前有0個執行緒在處理任務，還剩0個任務在佇列中等待，執行緒池的基本情況一路了然。

以上是Spring Boot怎麼使用執行緒池處理上萬個資料插入功能的詳細內容。更多資訊請關注PHP中文網其他相關文章！