ThreadPoolTaskExecutor使用

蔚落 2024-03-22 20:33 66阅读 0赞

### 1.ThreadPoolTaskExecutor和ThreadPoolExecutor区别 ###

ThreadPoolTaskExecutor是spring core包中的，而ThreadPoolExecutor是JDK中的JUC。

ThreadPoolTaskExecutor是对ThreadPoolExecutor进行了封装处理。

ThreadPoolExecutor结构，祖类都是调用Executor接口：

![4f426ba2c21f470e8897fe56c7e82b57.png][]

ThreadPoolTaskExecutor结构，祖类都是调用Executor接口：

![12feaae2f9174f92a726149f7ef3a2f1.png][]

### 2.工具类 ： Executors ###

Executors为线程池工具类，相当于一个工厂类，用来创建合适的线程池，返回ExecutorService类型的线程池。有如下方法：

ExecutorService newFixedThreadPool() : 创建固定大小的线程池

ExecutorService newCachedThreadPool() : 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量。

ExecutorService newSingleThreadExecutor() : 创建单个线程池。 线程池中只有一个线程

ScheduledExecutorService newScheduledThreadPool() : 创建可以执行延迟或定时任务的线程池

3.ThreadPoolTaskExecutor和ThreadPoolExecutor以及Executors到底该使用哪个？

ThreadPoolTaskExecutor和ThreadPoolExecutor比Executors创建线程池更加灵活，可以设置参数

推荐ThreadPoolTaskExecutor和ThreadPoolExecutor，而ThreadPoolTaskExecutor是ThreadPoolExecutor的封装，所以，性能更加优秀，**推荐ThreadPoolTaskExecutor**

4.这是ThreadPoolTaskExecutor用来初始化threadPoolExecutor的方法，BlockingQueue是一个阻塞队列，这个我们先不管。由于ThreadPoolTaskExecutor的实现方式完全是使用threadPoolExecutor进行实现，我们需要知道这个threadPoolExecutor的一些参数。

ThreadPoolExecutor的：

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements SchedulingTaskExecutor {
            private final Object poolSizeMonitor = new Object();
            private int corePoolSize = 1;
            private int maxPoolSize = 2147483647;
            private int keepAliveSeconds = 60;
            private boolean allowCoreThreadTimeOut = false;
            private int queueCapacity = 2147483647;
            private ThreadPoolExecutor threadPoolExecutor;   //这里就用到了ThreadPoolExecutor
    
    ThreadPoolTaskExecutor的：
    
         public ThreadPoolExecutor(int corePoolSize,
                                      int maximumPoolSize,
                                      long keepAliveTime,
                                      TimeUnit unit,
                                      BlockingQueue<Runnable> workQueue,
                                      ThreadFactory threadFactory,
                                      RejectedExecutionHandler handler) {
                if (corePoolSize < 0 ||
                    maximumPoolSize <= 0 ||
                    maximumPoolSize < corePoolSize ||
                    keepAliveTime < 0)
                    throw new IllegalArgumentException();
                if (workQueue == null || threadFactory == null || handler == null)
                    throw new NullPointerException();
                this.corePoolSize = corePoolSize;
                this.maximumPoolSize = maximumPoolSize;
                this.workQueue = workQueue;
                this.keepAliveTime = unit.toNanos(keepAliveTime);
                this.threadFactory = threadFactory;
                this.handler = handler;
            }

参数说明：

*  1、corePoolSize：核心线程数 （corePoolSize = cpu核心数 \* 2 + 有效磁盘数）
 *  2、queueCapacity：任务队列容量（阻塞队列）
 *  3、maxPoolSize：最大线程数
 *  4、 keepAliveTime：线程空闲时间
 *  5、allowCoreThreadTimeout：允许核心线程超时
 *  6、rejectedExecutionHandler：任务拒绝处理器

\* 两种情况会拒绝处理任务：  
\- 当线程数已经达到maxPoolSize，切队列已满，会拒绝新任务  
\- 当线程池被调用shutdown()后，会等待线程池里的任务执行完毕，再shutdown。如果在调用shutdown()和线程池真正shutdown之间提交任务，会拒绝新任务  
\* 线程池会调用rejectedExecutionHandler来处理这个任务。如果没有设置默认是AbortPolicy，会抛出异常  
\* ThreadPoolExecutor类有几个内部实现类来处理这类情况：  
\- AbortPolicy 直接抛出java.util.concurrent.RejectedExecutionException异常  
\- CallerRunsPolicy 若已达到待处理队列长度，将由主线程直接处理请求  
\- DiscardPolicy 抛弃当前任务；会导致被丢弃的任务无法再次被执行  
\- DiscardOldestPolicy 抛弃旧的任务；会导致被丢弃的任务无法再次被执行  
\* 实现RejectedExecutionHandler接口，可自定义处理器

### execute和submit的区别 ###

#### 提交任务类型 ####

*  execute和submitsubmitsubmit都属于线程池的方法，execute只能提交Runnable类型的任务
 *  submit既能提交Runnable类型任务也能提交Callable类型任务。

#### 返回值 ####

*  execute()没有返回值
 *  submit有返回值，所以需要返回值的时候必须使用submit

注意：submit的返回值为Future<?>,可通过get获取其泛型。Future<?>对象.get()方法具有阻塞当前线程的做用（也就是谁调用Future<?>对象.get()就阻塞谁，防止主线程在拿返回值时，而子线程还没有执行完）

### 线程池配置 ###

import org.springframework.context.annotation.Bean;
        import org.springframework.context.annotation.Configuration;
        import org.springframework.scheduling.annotation.EnableAsync;
        import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
        import java.util.concurrent.Executor;
        import java.util.concurrent.ThreadPoolExecutor;
         
        @Configuration
        public class ThreadPoolConfig {
         
            //获取cpu核心数
            private final static int AVAILABLE_PROCESSOR = Runtime.getRuntime().availableProcessors();
         
            /**
             * 通用线程池配置
             */
            @Bean("taskPoolExecutor")
            public Executor taskExecutor() {
                ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
                //设置核心线程数
                taskExecutor.setCorePoolSize(6);
                //设置最大线程数
                taskExecutor.setMaxPoolSize(6*2);
                //设置队列最大容量
                taskExecutor.setQueueCapacity(12);
                //设置线程空闲时间（秒）
                taskExecutor.setKeepAliveSeconds(60);
                //设置线程名称前缀
                taskExecutor.setThreadNamePrefix("taskPoolExecutor--");
                //调度器shutdown方法被调用时等待当前被调度的任务完成
                taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
                //等待时间（秒）
                taskExecutor.setAwaitTerminationSeconds(60);
                //拒绝策略
                taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
                //初始化线程池
                taskExecutor.initialize();
                return taskExecutor;
            }
         
             /**
             * 发送短信线程池
             * @return
             */
            @Bean("sendSmsThreadPool")
            public Executor sendSmsThreadPool() {
                ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
                //设置线程池参数信息
                taskExecutor.setCorePoolSize(AVAILABLE_PROCESSOR);
                taskExecutor.setMaxPoolSize(AVAILABLE_PROCESSOR + 1);
                taskExecutor.setQueueCapacity(256);
                taskExecutor.setKeepAliveSeconds(20);
                taskExecutor.setThreadNamePrefix("sendSmsThreadPool--");
                taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
                taskExecutor.setAwaitTerminationSeconds(60);
                //修改拒绝策略为使用当前线程执行
                taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
                //初始化线程池
                taskExecutor.initialize();
                return taskExecutor;
            }
         
            /**
             * 发送邮件线程池
             * @return
             */
            @Bean("sendEmailThreadPool")
            public Executor sendEmailThreadPool() {
                ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
                //设置线程池参数信息
                taskExecutor.setCorePoolSize(AVAILABLE_PROCESSOR);
                taskExecutor.setMaxPoolSize(AVAILABLE_PROCESSOR + 1);
                taskExecutor.setQueueCapacity(256);
                taskExecutor.setKeepAliveSeconds(20);
                taskExecutor.setThreadNamePrefix("sendEmailThreadPool--");
                taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
                taskExecutor.setAwaitTerminationSeconds(60);
                //修改拒绝策略为使用当前线程执行
                taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
                //初始化线程池
                taskExecutor.initialize();
                return taskExecutor;
            }
         
            @Bean(name = "threadPoolTaskExecutor")
            public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
                ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
                executor.setCorePoolSize(5);
                executor.setMaxPoolSize(5);
                executor.setQueueCapacity(10);
                executor.setKeepAliveSeconds(300);
                executor.setThreadNamePrefix("thread-ayokredit"); //线程名称
                executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
                return executor;
            }
               
        }

ThreadPoolTaskExecutor是Spring Framework中的一个类，用于在应用程序中执行异步任务。它使用Java中的线程池来管理任务的执行。使用方法如下:

#### 1、在配置文件中配置ThreadPoolTaskExecutor，包括线程池大小、队列大小等参数。 ####

import org.springframework.context.annotation.Configuration;
        import org.springframework.scheduling.annotation.EnableAsync;
        import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
         
        import java.util.concurrent.ThreadPoolExecutor;
         
        /**
         * 线程池配置
         */
        @Configuration
        @EnableAsync
        public class ThreadPoolConfig {
         
            // 核心线程数
            private int corePoolSize = 10;
            // 最大线程数
            private int maxPoolSize = 50;
            // 队列最大长度
            private int queueCapacity = 500;
            // 线程池维护线程所允许的空闲时间
            private int keepAliveSeconds = 300;
            // 线程名称
            private String threadNamePrefix = "TASK_EXECUTOR-";
         
            /**
             * 定长线程池
             *
             * @return
             */
            public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
                ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
                taskExecutor.setThreadNamePrefix(threadNamePrefix);
                taskExecutor.setMaxPoolSize(maxPoolSize);
                taskExecutor.setCorePoolSize(corePoolSize);
                taskExecutor.setQueueCapacity(queueCapacity);
                taskExecutor.setKeepAliveSeconds(keepAliveSeconds);
                // 拒绝策略
                taskExecutor.setRejectedExecutionHandler(new         ThreadPoolExecutor.CallerRunsPolicy());
                return taskExecutor;
            }
        }

#### 2、在需要执行异步任务的类中注入ThreadPoolTaskExecutor，调用ThreadPoolTaskExecutor的execute()方法来执行任务。 ####

@Autowired
        private ThreadPoolTaskExecutor taskExecutor;
         
        public void executeAsyncTask() {
            taskExecutor.execute(() -> {
                // 任务代码
            });
        }

[4f426ba2c21f470e8897fe56c7e82b57.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/04b1830b63c54cbba93da122164a94d0.png
[12feaae2f9174f92a726149f7ef3a2f1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/14/33d359f82c7b4c59b8aad7291549091d.png