java并发-线程池ThreadPoolTaskExecutor和ThreadPoolExecutor有何区别?

àì夳堔傛蜴生んèń 2022-12-20 03:06 171阅读 0赞

### 文章目录 ###

*   *  一、ThreadPoolTaskExecutor和ThreadPoolExecutor有何区别
     *  二、ThreadPoolTaskExecutor 线程池等待所有任务完成的几种方式
     *   *  方式1【推荐】: 使用 CountDownLatch
         *  方式2: 使用 Future
         *  方式3: setWaitForTasksToCompleteOnShutdown(true)
     *  三、ThreadPoolExecutor核心参数说明
     *  四、Spring使用ThreadPoolTaskExecutor自定义线程池及实现异步调用
     *  五、参考

## 一、ThreadPoolTaskExecutor和ThreadPoolExecutor有何区别 ##

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

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

> ThreadPoolTaskExecutor这个类则是spring包下的，是sring为我们提供的线程池类。

ThreadPoolExecutor结构，祖类都是调用Executor接口：  
![在这里插入图片描述][20201110120003946.png_pic_center]再来看一下ThreadPoolTaskExecutor结构，祖类都是调用Executor接口：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ludGhhdA_size_16_color_FFFFFF_t_70_pic_center]再来看一下源码：

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用来初始化threadPoolExecutor的方法，BlockingQueue是一个阻塞队列，这个我们先不管。由于**ThreadPoolTaskExecutor的实现方式完全是使用threadPoolExecutor进行实现**，我们需要知道这个threadPoolExecutor的一些参数。

## 二、ThreadPoolTaskExecutor 线程池等待所有任务完成的几种方式 ##

ThreadPoolTaskExecutor 线程池等待所有任务完成的几种方式  
参考URL: https://www.letianbiji.com/spring/spring-ThreadPoolTaskExecutor-wait-tasks-done.html

建议参考原文。

### 方式1【推荐】: 使用 CountDownLatch ###

public class TestThreadPool { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
            threadPoolTaskExecutor.setThreadNamePrefix("xxxx-");
            threadPoolTaskExecutor.setCorePoolSize(1);
            threadPoolTaskExecutor.setMaxPoolSize(5);
            threadPoolTaskExecutor.setQueueCapacity(5);
            threadPoolTaskExecutor.setKeepAliveSeconds(300);
            threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy() { 
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
                    System.out.println("丢弃");
                }
            });
            threadPoolTaskExecutor.initialize();
    
            int taskNum = 10;
            CountDownLatch latch =  new CountDownLatch(taskNum);
    
            for (int i=0; i < taskNum; i++) { 
                final int taskId = i;
                System.out.println("-- submit task " + taskId);
                threadPoolTaskExecutor.execute(() -> { 
                    try { 
                        System.out.printf("task: %s, thread: %s, start at %d\n", taskId, Thread.currentThread().getName(), System.currentTimeMillis()/1000);
                        Thread.sleep(2_000);
                        System.out.printf("task: %s, thread: %s, end at %d\n", taskId,  Thread.currentThread().getName(), System.currentTimeMillis()/1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        latch.countDown();
                    }
                });
            }
    
            latch.await();
            threadPoolTaskExecutor.shutdown();
        }
    }

### 方式2: 使用 Future ###

ThreadPoolTaskExecutor 线程池等待所有任务完成的几种方式  
参考URL: https://www.letianbiji.com/spring/spring-ThreadPoolTaskExecutor-wait-tasks-done.html  
建议参考原文。

### 方式3: setWaitForTasksToCompleteOnShutdown(true) ###

ThreadPoolTaskExecutor 线程池等待所有任务完成的几种方式  
参考URL: https://www.letianbiji.com/spring/spring-ThreadPoolTaskExecutor-wait-tasks-done.html

建议参考原文。

## 三、ThreadPoolExecutor核心参数说明 ##

ThreadPoolExecutor参数说明  
参考URL: https://blog.csdn.net/sinat\_15946141/article/details/107951917

一、ThreadPoolExecutor核心参数说明

1、corePoolSize：核心线程数
            * 核心线程会一直存活，及时没有任务需要执行
            * 当线程数小于核心线程数时，即使有线程空闲，线程池也会优先创建新线程处理
            * 设置allowCoreThreadTimeout=true（默认false）时，核心线程会超时关闭
     
        2、queueCapacity：任务队列容量（阻塞队列）
            * 当核心线程数达到最大时，新任务会放在队列中排队等待执行
     
        3、maxPoolSize：最大线程数
            * 当线程数>=corePoolSize，且任务队列已满时。线程池会创建新线程来处理任务
            * 当线程数=maxPoolSize，且任务队列已满时，线程池会拒绝处理任务而抛出异常
     
        4、 keepAliveTime：线程空闲时间
            * 当线程空闲时间达到keepAliveTime时，线程会退出，直到线程数量=corePoolSize
            * 如果allowCoreThreadTimeout=true，则会直到线程数量=0
     
        5、allowCoreThreadTimeout：允许核心线程超时
        6、rejectedExecutionHandler：任务拒绝处理器
            * 两种情况会拒绝处理任务：
                - 当线程数已经达到maxPoolSize，切队列已满，会拒绝新任务
                - 当线程池被调用shutdown()后，会等待线程池里的任务执行完毕，再shutdown。如果在调用shutdown()和线程池真正shutdown之间提交任务，会拒绝新任务
            * 线程池会调用rejectedExecutionHandler来处理这个任务。如果没有设置默认是AbortPolicy，会抛出异常
            * ThreadPoolExecutor类有几个内部实现类来处理这类情况：
                - AbortPolicy 丢弃任务，抛运行时异常
                - CallerRunsPolicy 执行任务
                - DiscardPolicy 忽视，什么都不会发生
                - DiscardOldestPolicy 从队列中踢出最先进入队列（最后一个执行）的任务
            * 实现RejectedExecutionHandler接口，可自定义处理器

二、ThreadPoolExecutor执行顺序

线程池按以下行为执行任务
        1. 当线程数小于核心线程数时，创建线程。
        2. 当线程数大于等于核心线程数，且任务队列未满时，将任务放入任务队列。
        3. 当线程数大于等于核心线程数，且任务队列已满
            -1 若线程数小于最大线程数，创建线程
            -2 若线程数等于最大线程数，抛出异常，拒绝任务

三、ThreadPoolExecutor如何设置参数

1、默认值
            * corePoolSize=1
            * queueCapacity=Integer.MAX_VALUE
            * maxPoolSize=Integer.MAX_VALUE
            * keepAliveTime=60s
            * allowCoreThreadTimeout=false
            * rejectedExecutionHandler=AbortPolicy()
     
        2、如何来设置
            * 需要根据几个值来决定
                - tasks ：每秒的任务数，假设为1000
                - taskcost：每个任务花费时间，假设为0.1s
                - responsetime：系统允许容忍的最大响应时间，假设为1s
            * 做几个计算
                - corePoolSize = 每秒需要多少个线程处理？ 
                    * 一颗CPU核心同一时刻只能执行一个线程，然后操作系统切换上下文，核心开始执行另一个线程的代码，以此类推，超过cpu核心数，就会放入队列，如果队列也满了，就另起一个新的线程执行，所有推荐：corePoolSize = ((cpu核心数 * 2) + 有效磁盘数)，java可以使用Runtime.getRuntime().availableProcessors()获取cpu核心数
                - queueCapacity = (coreSizePool/taskcost)*responsetime
                    * 计算可得 queueCapacity = corePoolSize/0.1*1。意思是队列里的线程可以等待1s，超过了的需要新开线程来执行
                    * 切记不能设置为Integer.MAX_VALUE，这样队列会很大，线程数只会保持在corePoolSize大小，当任务陡增时，不能新开线程来执行，响应时间会随之陡增。
                - maxPoolSize = (max(tasks)- queueCapacity)/(1/taskcost)
                    * 计算可得 maxPoolSize = (1000-corePoolSize)/10，即(每秒并发数-corePoolSize大小) / 10
                    * （最大任务数-队列容量）/每个线程每秒处理能力 = 最大线程数
                - rejectedExecutionHandler：根据具体情况来决定，任务不重要可丢弃，任务重要则要利用一些缓冲机制来处理
                - keepAliveTime和allowCoreThreadTimeout采用默认通常能满足

根据上面算出了核心的线程数corePoolSize，然后maxPoolSize和它相等，队列大小为上面公式计算。

1.计算密集型线程池

cpu使用率较高（也就是一些复杂运算，逻辑处理），所以线程数一般只需要cpu核数的线程就可以了。 这一类型的在开发中多出现的一些业务复杂计算和逻辑处理过程中。

数量一般为 N+1个 N为CPU核心数

2.IO密集型

cpu使用率较低，程序中会存在大量I/O操作占据时间，导致线程空余时间出来，所以通常就需要开cpu核数的两倍的线程， 当线程进行I/O操作cpu空暇时启用其他线程继续使用cpu，提高cpu使用率 通过上述可以总结出：线程的最佳数量： 最佳线程数目 = （（线程等待时间+线程CPU时间）/线程CPU时间 ）\* CPU数目 线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程。这一类型在开发中主要出现在一些读写操作频繁的业务逻辑中。

数量一般为：2N + 1个 N为CPU核心数

## 四、Spring使用ThreadPoolTaskExecutor自定义线程池及实现异步调用 ##

Spring使用ThreadPoolTaskExecutor自定义线程池及实现异步调用  
参考URL: https://segmentfault.com/a/1190000020674056

## 五、参考 ##

JAVA线程池学习，ThreadPoolTaskExecutor和ThreadPoolExecutor有何区别?  
参考URL: https://cloud.tencent.com/developer/article/1408125

[20201110120003946.png_pic_center]: /images/20221120/b917e989a59c48dbab822f60fc35413c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ludGhhdA_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/202011101200167.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ludGhhdA==,size_16,color_FFFFFF,t_70#pic_center