JUC高并发编程

JUC高并发编程 十一、ThreadPool 线程池 11.1）线程池简介

线程池（英语：thread pool）：一种线程使用模式。线程过多会带来调度开销， 进而影响缓存局部性和整体性能；而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务；这避免了在处理短时间任务时创建与销毁线程的代价。

程池不仅能够保证内核的充分利用，还能防止过分调度。

11.1.1）线程池简介

线程池做的工作只要是控制运行的线程数量，处理过程中将任务放入队列，然后在线程创建后启动这些任务，如果线程数量超过了最大数量， 超出数量的线程排队等候，等其他线程执行完毕，再从队列中取出任务来执行。

11.1.2）线程池特点

线程池的主要特点为：

• 降低资源消耗：通过重复利用已创建的线程降低线程创建和销毁造成的销耗；

• 提高响应速度：当任务到达时，任务可以不需要等待线程创建就能立即执行；

• 提高线程的可管理性：线程是稀缺资源，如果无限制的创建，不仅会销耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控；

• Java 中的线程池是通过 Executor 框架实现的，该框架中用到了 Executor，Executors， ExecutorService，ThreadPoolExecutor这几个类 ，其关系如下图：

 

11.2）线程池参数说明

常用参数：

• corePoolSize：线程池的常驻线程数量(核心)

• maximumPoolSize：线程池能容纳的最大线程数

• keepAliveTime空闲：线程存活时间

• unit ：线程存活的时间单位

• BlockingQueue workQueue ：存放提交但未执行任务的队列 【阻塞队列】

• threadFactory： 创建线程的工厂类【线程工厂】

• handler ：等待队列满后的拒绝策略

线程池中，有三个重要的参数，决定影响了拒绝策略：

corePoolSize - 核心线程数，也即最小的线程数；

workQueue - 阻塞队列；

maximumPoolSize - 最大线程数

当提交任务数大于 corePoolSize 的时候，会优先将任务放到 workQueue 阻塞队列中；当阻塞队列饱和后，会扩充线程池中线程数，直到达到 maximumPoolSize 最大线程数配置。此时再多余的任务，则会触发线程池的拒绝策略。

总结：当提交的任务数大于（workQueue.size() + maximumPoolSize ），就会触发线程池的拒绝策略

11.3）线程池底层工作原理

线程池底层工作原理如下图所示：

 线程池底层工作流程：

1. 在创建了线程池后，线程池中的线程数为零；

2. 当调用execute()方法添加一个请求任务时，线程池会做出如下判断：

   2.1） 如果正在运行的线程数量小于corePoolSize，那么马上创建线程运行这个任务；

   2.2）如果正在运行的线程数量大于或等于corePoolSize，那么将这个任务放入队列；

   2.3 ）如果这个时候队列满了且正在运行的线程数量还小于 maximumPoolSize，那么还是要创建非核心线程立刻运行这个任务；

   2.4 ）如果队列满了且正在运行的线程数量大于或等于maximumPoolSize，那么线程池会启动饱和拒绝策略来执行。

3. 当一个线程完成任务时，它会从队列中取下一个任务来执行；

4. 当一个线程无事可做超过一定的时间（keepAliveTime）时，线程会判断：

   4.1）如果当前运行的线程数大于corePoolSize，那么这个线程就被停掉；

   4.2）所以线程池的所有任务完成后，它最终会收缩到corePoolSize的大小

11.4）线程池的拒绝策略

JDK内置的拒绝策略如下图所示：

 说明：

1. AbortPolicy：线程池默认的拒绝策略；丢弃任务并抛出拒绝执行 RejectedExecutionException 异常信息；必须处理好抛出的异常，否则会打断当前的执 流程，影响后续的任务执行；

2. CallerRunsPolicy：当触发拒绝策略，只要线程池没有关闭的话，则使用调用线程直接运行任务，将任务退回给调用者；一般并发比较小，性能要求不高，不允许失败。但是由于调用者自己运行任务，如果任务提交速度过快，可能导致程序阻塞，性能损失；

3. DiscardOldestPolicy: 当触发拒绝策略，只要线程池没有关闭的话，丢弃阻塞队列 workQueue 中最老的一个任务，并将新任务加入提交；

4. DiscardPolicy: 直接丢弃，其他啥都不做

11.5）线程池使用方式

11.5.1）newFixedThreadPool

11.5.1.1）作用

【一池N线程】创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程。

在任意点，在大多数线程会处于处理任务的活动状态；如果在所有线程处于活动状态时提交附加任务，则在有可用线程之前，附加任务将在队列中等待。

如果在关闭前的执行期间由于失败而导致任何线程终止，那么一个新线 程将代替它执行后续的任务（如果需要），在某个线程被显式地关闭之前，池中的线程将一直存在。

11.5.1.2）特征

• 线程池中的线程处于一定的量，可以很好的控制线程的并发量；

• 线程可以重复被使用，在显示关闭之前，都将一直存在 ；

• 超出一定量的线程被提交时候需在队列中等待。

11.5.1.3）适用场景

适用于可以预测线程数量的业务中，或者服务器负载较重，对线程数有严格限制的场景

11.5.1.4）代码示例

//演示newFixedThreadPool创建线程池【一池N线程】
public class ThreadPoolDemo1 {
    public static void main(String[] args) {
        //一池N线程
        //5个窗口
        ExecutorService threadPool1 = Executors.newFixedThreadPool(5);
        //10个顾客请求
        try {
            for (int i = 1; i <= 10; i++) {
                //执行
                threadPool1.execute(() -> {
                    System.out.println("当前线程名：" + Thread.currentThread().getName() + " 办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭
            threadPool1.shutdown();
        }
    }
}

输出：

当前线程名：pool-1-thread-1 办理业务
当前线程名：pool-1-thread-5 办理业务
当前线程名：pool-1-thread-1 办理业务
当前线程名：pool-1-thread-4 办理业务
当前线程名：pool-1-thread-1 办理业务
当前线程名：pool-1-thread-2 办理业务
当前线程名：pool-1-thread-3 办理业务
当前线程名：pool-1-thread-1 办理业务
当前线程名：pool-1-thread-4 办理业务
当前线程名：pool-1-thread-5 办理业务

11.5.2） newSingleThreadExecutor

11.5.2.1）作用

【一池一线程】创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程，可保证顺序地执行各个任务，并且在任意给定的时间不会有多个线程是活动的。

与其他等效的 newFixedThreadPool不同，可保证无需重新配置此方法所返回的执行程序即 可使用其他的线程。

（注意，如果因为在关闭前的执行期间出现失败而终止了此单个线程， 那么如果需要，一个新线程将代替它执行后续的任务）。

一个任务一个任务执行，一池一线程

11.5.2.2）特征

线程池中最多执行1个线程，之后提交的线程活动将会排在队列中以此执行

11.5.2.3）特征

适用于需要保证顺序执行各个任务，并且在任意时间点，不会同时有多个 线程的场景

11.5.2.4）代码示例

//演示newSingleThreadExecutor创建线程池【一池一线程】
public class ThreadPoolDemo1 {
    public static void main(String[] args) {
        //一池一线程
        //一个窗口
        ExecutorService threadPool2 = Executors.newSingleThreadExecutor();
        //10个顾客请求
        try {
            for (int i = 1; i <= 10; i++) {
                //执行
                threadPool2.execute(() -> {
                    System.out.println("当前线程名：" + Thread.currentThread().getName() + " 办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭
            threadPool2.shutdown();
        }
    }
}

输出：

当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务
当前线程名：pool-2-thread-1 办理业务

11.5.3） newCachedThreadPool

11.5.3.1）作用

【一池可扩容线程】创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

线程池根据需求创建线程，可扩容，遇强则强

11.5.3.2）特点

• 线程池中数量没有固定，可达到最大值（Interger. MAX_VALUE）；

• 线程池中的线程可进行缓存重复利用和回收（回收默认时间为1分钟）；

• 当线程池中，没有可用线程，会重新创建一个线程

11.5.3.3）适用场景

适用于创建一个可无限扩大的线程池，服务器负载压力较轻，执行时间较 短，任务多的场景

11.5.3.4）代码示例

//演示newCachedThreadPool创建线程池【一池可扩容线程】
public class ThreadPoolDemo1 {
    public static void main(String[] args) {
        //一池可扩容线程
        ExecutorService threadPool3 = Executors.newCachedThreadPool();
        //10个顾客请求
        try {
            for (int i = 1; i <= 10; i++) {
                //执行
                threadPool3.execute(() -> {
                    System.out.println("当前线程名：" + Thread.currentThread().getName() + " 办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭
            threadPool3.shutdown();
        }
    }
}

输出：

当前线程名：pool-3-thread-2 办理业务
当前线程名：pool-3-thread-5 办理业务
当前线程名：pool-3-thread-3 办理业务
当前线程名：pool-3-thread-7 办理业务
当前线程名：pool-3-thread-4 办理业务
当前线程名：pool-3-thread-1 办理业务
当前线程名：pool-3-thread-8 办理业务
当前线程名：pool-3-thread-6 办理业务
当前线程名：pool-3-thread-10 办理业务
当前线程名：pool-3-thread-9 办理业务

11.6）自定义线程池

11.6.1）自定义线程池的原因

项目中创建多线程时，使用常见的三种线程池创建方式，newFixedThreadPool【一池N线程】、newSingleThreadExecutor【一池一线程】、newCachedThreadPool【一池可扩容线程】都有一定问题，原因是FixedThreadPool和SingleThreadExecutor底层都是用 LinkedBlockingQueue实现的，这个队列最大长度为Integer.MAX_VALUE，容易导致OOM。

为什么不允许适用不允许Executors.的方式手动创建线程池，如下图 ：

 所以实际生产一般自己通过ThreadPoolExecutor的7个参数来自定义线程池

11.6.2）创建自定义线程池

创建线程池推荐适用ThreadPoolExecutor及其7个参数手动创建：

• corePoolSize：线程池的核心线程数；

• maximumPoolSize：能容纳的最大线程数；

• keepAliveTime：空闲线程存活时间；

• unit：存活的时间单位；

• workQueue：存放提交但未执行任务的队列；

• threadFactory：创建线程的工厂类；

• handler：等待队列满后的拒绝策略

11.6.3）自定义线程池代码示例

自定义线程池代码示例：

//自定义线程池创建
public class ThreadPoolDemo2 {
    public static void main(String[] args) {
        // 自定义线程池
        ExecutorService threadPool = new ThreadPoolExecutor(
                2,
                5,
                2L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
​
        //10个顾客请求
        try {
            for (int i = 1; i <= 10; i++) {
                //执行
                threadPool.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + " 办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //关闭
            threadPool.shutdown();
        }
    }
}

输出：

pool-1-thread-2 办理业务
pool-1-thread-5 办理业务
pool-1-thread-5 办理业务
pool-1-thread-4 办理业务
pool-1-thread-1 办理业务
pool-1-thread-3 办理业务
pool-1-thread-5 办理业务
pool-1-thread-2 办理业务
java.util.concurrent.RejectedExecutionException: Task com.study.pool.ThreadPoolDemo2$$Lambda$1/1096979270@7ba4f24f rejected from java.util.concurrent.ThreadPoolExecutor@3b9a45b3[Running, pool size = 5, active threads = 3, queued tasks = 0, completed tasks = 5]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
    at com.study.pool.ThreadPoolDemo2.main(ThreadPoolDemo2.java:23)