spring自定义线程池逻辑配置 ThreadPoolTaskExecutor corePoolSize maxPoolSize queueCapacity rejectedExecutionHa

本是古典何须时尚 2024-05-23 20:32 69阅读 0赞

## spring自定义线程池 逻辑 配置 ThreadPoolTaskExecutor corePoolSize maxPoolSize queueCapacity rejectedExecutionHa ##

### 线程池简介 ###

*  线程池是一个创建、使用、销毁线程的调度中心，保证线程充分利用，也能防止过分调度。
 *  线程池对于异步任务执行，提高并发能力非常友好。线程池的最大优点是快速响应任务执行，重复利用线程，避免频繁创建和销毁线程。
 *  对于java的spring框架，支持 1.自定义线程池、2.使用spring默认线程池 3.重写spring默认线程池

### spring线程池配置参数 ###

*  ![在这里插入图片描述][87f53157261e4378892b66d2a37bdb06.png]

### 线程池调用逻辑 ###

*  corePoolSize和queueCapacity，再程序运行时，是一个`动态的值`。
 *  当应用程序运行起来时，线程池不会自动创建corePoolSize的数量的空闲线程，而是线程池被调用时，再去一个一个创建线程。
 *  当线程池已有的线程数量未达到corePoolSize数量时，且线程都不是空闲中，新的任务进来需要执行，此时会再创建新线程。
 *  当线程池已有的线程数量达到corePoolSize数量时，且线程存在空闲中，新的任务进来需要执行，此时会直接复用空闲线程。
 *  当线程池已有的线程数量达到corePoolSize数量时，且线程都不是空闲中，新的任务进来需要执行，此时会再创建新线程，但不会超过maxPoolSize的值。
 *  当线程池已有的线程数量达到maxPoolSize数量时，不会再创建新线程，但是会将任务放入任务队列中，等待空闲的线程。
 *  当线程池中任务队列数量达到queueCapacity数量，或awaitTerminationSeconds任务等待时间超出后，会触发终止策略rejectedExecutionHandler的执行, 默认是抛出异常RejectedExecutionException。
 *  任务执行完成后，线程的销毁，**先会销毁超过corePoolSize数量的`多余`线程**，继续保留corePoolSize数量的空闲线程，空闲线程`需要等待活跃时间`（keepAliveSeconds）超过后再销毁。
 *  线程池中任务队列采用BlockingQueue， 阻塞队列，先进先出。  
    当线程执行完任务空闲后，线程池从任务队列中，拿取最旧任务，交给空闲线程去执行， queueCapacity相对会减1。
 *  如果任务队列没有任务，且也没有新的任务进来，进入空闲状态60秒（keepAliveSeconds配置60秒），60秒过程中，一直没有任务需要被线程去执行，这时线程被销毁，corePoolSize数值会减1， 直到变成0。
 *  当任务需要被线程执行时，首先会判断是否有空闲线程，表明有空闲线程，可以直接复用线程，不用新创建。 这个过程有线程池自己管理。 通过线程池对象的getActiveCount()小于getPoolSize()是 可以判断有空闲线程。

### 怎样确定数值corePoolSize、maxPoolSize、queueCapacity等 ###

*  对于应用程序，对于不同的业务，可能需要不同的线程池来满足。  
    像corePoolSize、maxPoolSize、queueCapacity这类参数需要考虑硬件机器，业务功能来调优，也可能在应用程序的运行时来动态修改参数。这是一个调优的过程。
 *  举例一个订单系统的业务
 *  要求：订单接收必须要快（1秒钟）, 订单流程处理可以相对滞后（多处理个十几秒）
 *  可以定义2套线程池，
 *  订单接收功能，corePoolSize、maxPoolSize、queueCapacity 数值都稍微放大，资源尽量给到订单接收，提高吞请求吐量
 *  订单流程处理功能，corePoolSize、maxPoolSize、queueCapacity 数值都稍微放小。

### 代码实现过程 ###

*  下面代码示例，以不同业务，定义三个线程池为示例：
 *  1.普通业务线程池、2.订单流程处理功能 3.订单接收功能

#### 应用程序启动类，添加@EnableAsync， 开启异步线程 ####

#### 定义线程池的名称 ####

*      package com.xxxxx.config;
        
        public class ThreadPoolNameKey {
        
            /**
             * 默认线程池
             */
            public final static String customDefaultThreadPool = "customDefaultThreadPool";
        
            /**
             * 接收订单业务
             */
            public final static String receiveOrderThreadPool = "receiveOrderThreadPool";
        
            /**
             * 处理订单业务
             */
            public final static String handlerOrderThreadPool = "handlerOrderThreadPool";
        }

#### 定义线程池相关参数 ####

*      package com.xxxxx.config;
        
        import lombok.Data;
        import lombok.NoArgsConstructor;
        
        @Data
        @NoArgsConstructor
        public class ThreadPoolConfig {
            /**
             * 核心线程池大小
             */
            private int corePoolSize = 1;
        
            /**
             * 线程池中最大线程数
             */
            private int maxPoolSize = Integer.MAX_VALUE;
        
            /**
             * 线程池中线程的空闲存活时间, 单位：秒
             */
            private int keepAliveSeconds = 60;
        
            /**
             * 任务队列数量
             */
            private int queueCapacity = Integer.MAX_VALUE;
        
            /**
             * 设置线程池中任务的等待时间, 单位：秒
             */
            private int awaitTerminationSeconds = 180;
        
            /**
             * 线程池名的前缀
             */
            private String threadNamePrefix = "ThreadPoolConfig";
        
            /**
             * 允许核心线程超时
             */
            private boolean allowCoreThreadTimeOut = true;
        
            public ThreadPoolConfig(int corePoolSize, int maxPoolSize, String threadNamePrefix){
                this.corePoolSize = corePoolSize;
                this.maxPoolSize = maxPoolSize;
                this.threadNamePrefix = threadNamePrefix;
            }
        }
 *  数值可以配置在yml文件中，首先在应用程序启动类，添加@EnableConfigurationProperties，  
    再在ThreadPoolConfig上面添加注解@ConfigurationProperties(), 指定yml配置项

#### 实现线程池配置 ####

*      package com.xxxxx.config;
        
        import org.springframework.context.annotation.Bean;
        import org.springframework.context.annotation.Configuration;
        import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
        
        import java.util.concurrent.Executor;
        
        @Configuration
        public class ThreadPool {
        
            /**
             * 默认线程池
             * @return
             */
            @Bean(name = ThreadPoolNameKey.customDefaultThreadPool)
            public Executor customDefaultThreadPool(){
                ThreadPoolConfig config = new ThreadPoolConfig(10, 20, ThreadPoolNameKey.customDefaultThreadPool);
                return this.customThreadPool(config);
            }
        
            /**
             * 接收订单业务-线程池
             * @return
             */
            @Bean(name = ThreadPoolNameKey.receiveOrderThreadPool)
            public Executor receiveOrderThreadPool(){
                ThreadPoolConfig config = new ThreadPoolConfig(20, 50, ThreadPoolNameKey.receiveOrderThreadPool);
                return this.customThreadPool(config);
            }
        
            /**
             * 处理订单业务-线程池
             * @return
             */
            @Bean(name = ThreadPoolNameKey.handlerOrderThreadPool)
            public Executor handlerOrderThreadPool(){
                ThreadPoolConfig config = new ThreadPoolConfig(10, 20, ThreadPoolNameKey.handlerOrderThreadPool);
                return this.customThreadPool(config);
            }
        
            
            /**
             * 构造线程池
             * @return
             */
            private Executor customThreadPool(ThreadPoolConfig config) {
                ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
                //核心线程池大小
                executor.setCorePoolSize(config.getCorePoolSize());
                //线程池中最大线程数
                executor.setMaxPoolSize(config.getMaxPoolSize());
                //线程池中线程的空闲存活时间
                executor.setKeepAliveSeconds(config.getKeepAliveSeconds());
                //任务队列数量；线程数达到最大时，新任务会放在队列中排队等待执行。
                executor.setQueueCapacity(config.getQueueCapacity());
                //允许核心线程超时
                executor.setAllowCoreThreadTimeOut(config.isAllowCoreThreadTimeOut());
                //线程池名的前缀
                executor.setThreadNamePrefix(config.getThreadNamePrefix());
                //是否等待任务完成再结束线程池
                executor.setWaitForTasksToCompleteOnShutdown(Boolean.TRUE);
                //设置线程池中任务的等待时间，如果超过这个时间中断正在进行的任务并清除，确保任务都能被关闭，而不是阻塞住。
                executor.setAwaitTerminationSeconds(config.getAwaitTerminationSeconds());
        
                //任务调用线程间传递数据，或者为任务执行提供一些监控/统计信息
                //executor.setTaskDecorator();
        
                //线程工厂
                //executor.setThreadFactory();
        
                //Bean名称, 因为已经@Bean(name = "xxx"), 索引可不填
                //executor.setBeanName();
        
                //线程工厂的线程的优先级
                //executor.setThreadPriority();
        
                //是否应该创建守护线程
                //executor.setDaemon();
        
                //创建线程的线程组。
                //executor.setThreadGroup();
        
                //线程组名称
                //executor.setThreadGroupName();
        
                //拒绝策略
                //AbortPolicy()：默认值，直接抛出异常。
                //CallerRunsPolicy()：由调用的主线程执行该任务
                //DiscardPolicy()：直接丢弃任务
                //DiscardOldestPolicy()：丢弃线程队列最旧的未处理任务，添加新任务
                //executor.setRejectedExecutionHandler();
        
                executor.initialize();
                return executor;
            }
        }

#### 业务异步执行，使用相对应的线程池 ####

*      package com.xxxxx.component;
        
        import com.pd.shopping.order.config.ThreadPoolNameKey;
        import org.apache.commons.lang.time.DateFormatUtils;
        import org.springframework.scheduling.annotation.Async;
        import org.springframework.stereotype.Component;
        
        import java.util.Date;
        
        @Component
        public class OrderComponent {
        
            @Async(value = ThreadPoolNameKey.customDefaultThreadPool)
            public void testDefaultThreadPool(Integer i){
                try {
                    System.out.print(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + " >> 默认线程名称-开始：" + "索引-" + i + " >> " + Thread.currentThread().getName() + "\n");
                    Thread.sleep(1000L);
                    System.out.print(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + " >> 默认线程名称-结束：" + "索引-" + i + " >> " + Thread.currentThread().getName() + "\n");
                } catch (Exception e){
                    System.out.print("睡眠线程异常：" +  e.toString());
                }
            }
        
            @Async(value = ThreadPoolNameKey.receiveOrderThreadPool)
            public void testReceiveOrderThreadPool(Integer i){
                try {
                    System.out.print(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + " >> 接单线程名称-开始：" + "索引-" + i + " >> " + Thread.currentThread().getName() + "\n");
                    Thread.sleep(1000L);
                    System.out.print(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + " >> 接单线程名称-结束：" + "索引-" + i + " >> " + Thread.currentThread().getName() + "\n");
                } catch (Exception e){
                    System.out.print("睡眠线程异常：" +  e.toString());
                }
            }
        
            @Async(value = ThreadPoolNameKey.handlerOrderThreadPool)
            public void testHandlerOrderThreadPool(Integer i){
                try {
                    System.out.print(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + " >> 处理订单线程名称-开始：" + "索引-" + i + " >> " + Thread.currentThread().getName() + "\n");
                    Thread.sleep(1000L);
                    System.out.print(DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss") + " >> 处理订单线程名称-结束：" + "索引-" + i + " >> " + Thread.currentThread().getName() + "\n");
                } catch (Exception e){
                    System.out.print("睡眠线程异常：" +  e.toString());
                }
            }
        }

#### 测试调用 ####

*  获取线程池对象， 可在运行时动态修改线程池部分参数
 *      /**
          * 注入线程池，来获取线程池对象
          * 
          * 自定义线程池名称 customDefaultThreadPool、receiveOrderThreadPool、handlerOrderThreadPool
          * 
          * 通过变量名称，可以自动匹配 @Bean("customDefaultThreadPool") 这个线程池
          */
        @Lazy
        @Autowired
        private ThreadPoolTaskExecutor customDefaultThreadPool;
 *  ***@Lazy延时注入，主要防止应用程序启动时，@Configuration还没完成，导致获取线程池对象失败***。
 *  customDefaultThreadPool对象中中可以重新set部分参数。

#### 测试调用 ####

*      @Lazy
        @Autowired
        private ThreadPoolTaskExecutor customDefaultThreadPool;
        
        
        public String testThreadPool(int forEachNum) {
            int activeCount = customDefaultThreadPool.getActiveCount();
            int corePoolSize = customDefaultThreadPool.getPoolSize();
        
            System.out.print("customDefault activeCount：" +  activeCount + "\n");
            System.out.print("customDefault corePoolSize：" +  corePoolSize + "\n");
        
            for (Integer i = 1 ; i <= forEachNum; i ++){
                try {
                    orderComponent.testDefaultThreadPool(i);
                    //orderComponent.testReceiveOrderThreadPool(i);
                    //orderComponent.testHandlerOrderThreadPool(i);
                } catch (RejectedExecutionException e){
                    return "线程资源不够用, 拒绝执行任务，请稍后重试";
                } catch (Exception e){
                    return "";
                }
            }
            return "success";
        }

[87f53157261e4378892b66d2a37bdb06.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/b69b2f865d2e4b13ade7bc5d6e2680c5.png