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

一、ThreadPoolTaskExecutor和ThreadPoolExecutor有何区别

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

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

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

ThreadPoolExecutor结构，祖类都是调用Executor接口：
再来看一下ThreadPoolTaskExecutor结构，祖类都是调用Executor接口：

再来看一下源码：

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
1
2
3
4
5
6
7
8

这是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();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

方式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接口，可自定义处理器
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

二、ThreadPoolExecutor执行顺序

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

三、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采用默认通常能满足
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

根据上面算出了核心的线程数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