并发编程线程池及ThreadPoolExecutor类的原理解析(二十七)

末蓝、 2022-12-31 02:28 182阅读 0赞

今天我们开始对并发编程线程池及ThreadPoolExecutor类的原理解析：

**为什么要用线程池？**

Java  中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行 任务的程序都可以使用线程池。在开发过程中，合理地使用线程池能够带来 3 个好处。

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

第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立 即执行。假设一个服务器完成一项任务所需时间为：T1  创建线程时间， T2  在线 程中执行任务的时间，T3  销毁线程时间。 如果：T1 + T3  远大于  T2 ，则可以 采用线程池，以提高服务器性能。线程池技术正是关注如何缩短或调整 T1,T3  时 间的技术，从而提高服务器程序性能的。它把 T1 ， T3  分别安排在服务器程序的 启动和结束的时间段或者一些空闲的时间段，这样在服务器程序处理客户请求时， 不会有 T1 ， T3  的开销了。

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

假设一个服务器一天要处理  50000  个请求，并且每个请求需要一个单独的线 程完成。在线程池中，线程数一般是固定的，所以产生线程总数不会超过线程池 中线程的数目，而如果服务器不利用线程池来处理这些请求则线程总数为 50000 。 一般线程池大小是远小于 50000 。所以利用线程池的服务器程序不会为了创建 50000 而在处理请求时浪费时间，从而提高效率。

线程池的原理图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

**ThreadPoolExecutor**  **的类关系**

Executor  是一个接口，它是  Executor  框架的基础，它将任务的提交与任务的 执行分离开来。 ExecutorService 接口继承了  Executor ，在其上做了一些  shutdown() 、 submit() 的扩展，可以说是真正的线程池接口； AbstractExecutorService 抽象类实现了  ExecutorService  接口中的大部分方法； ThreadPoolExecutor 是线程池的核心实现类，用来执行被提交的任务。 ScheduledExecutorService 接口继承了  ExecutorService  接口，提供了带 " 周期 执行" 功能  ExecutorService ； ScheduledThreadPoolExecutor是一个实现类，可以在给定的延迟后运行命令， 或者定期执行命令。ScheduledThreadPoolExecutor  比  Timer  更灵活，功能更强大。

线程池的结构图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

**线程池的创建各个参数含义**

创建线程池有多种构造方法：

一种构造方法源码：

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  RejectedExecutionHandler handler) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), handler);
        }

创建一个线程池的工具类：

public class ThreadPool {
    
        private static ThreadPoolExecutor executor;
    
        public static ThreadPoolExecutor getPool(){
            if(executor == null || executor.isTerminated()){
                synchronized (ThreadPool.class){
                    if(executor == null || executor.isTerminated()){
                        executor = new ThreadPoolExecutor(10,16,
                                60, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(100),
                                new ThreadPoolExecutor.CallerRunsPolicy());
                        executor.allowCoreThreadTimeOut(true);
                    }
                }
            }
            return executor;
        }
    }

此篇以下面这个构造方法讲解：

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 threadFactory, defaultHandler);
        }

**corePoolSize**

线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任 务，直到当前线程数等于 corePoolSize ； 如果当前线程数为 corePoolSize ，继续提交的任务被保存到阻塞队列中，等 待被执行； 如果执行了线程池的 prestartAllCoreThreads() 方法，线程池会提前创建并启 动所有核心线程。

**maximumPoolSize**

线程池中允许的最大线程数。如果当前阻塞队列满了，且继续提交任务，则 创建新的线程执行任务，前提是当前线程数小于 maximumPoolSize

**keepAliveTime**

线程空闲时的存活时间，即当线程没有任务执行时，继续存活的时间。默认 情况下，该参数只在线程数大于 corePoolSize  时才有用

**TimeUnit**

keepAliveTime  的时间单位

**workQueue**

workQueue  必须是  BlockingQueue  阻塞队列。当线程池中的线程数超过它的 corePoolSize 的时候，线程会进入阻塞队列进行阻塞等待。通过  workQueue ，线 程池实现了阻塞功能

用于保存等待执行的任务的阻塞队列，一般来说，我们应该尽量使用有界队 列，因为使用无界队列作为工作队列会对线程池带来如下影响。

1 ）当线程池中的线程数达到  corePoolSize  后，新任务将在无界队列中等待， 因此线程池中的线程数不会超过 corePoolSize 。

2 ）由于  1 ，使用无界队列时  maximumPoolSize  将是一个无效参数。

3 ）由于  1  和  2 ，使用无界队列时  keepAliveTime  将是一个无效参数。

4 ）更重要的，使用无界  queue  可能会耗尽系统资源，有界队列则有助于防 止资源耗尽，同时即使使用有界队列，也要尽量控制队列的大小在一个合适的范 围。

所以我们一般会使用， ArrayBlockingQueue 、 LinkedBlockingQueue 、 SynchronousQueue、 PriorityBlockingQueue 。

**threadFactory**

创建线程的工厂，通过自定义的线程工厂可以给每个新建的线程设置一个具 有识别度的线程名，当然还可以更加自由的对线程做更多的设置，比如设置所有 的线程为守护线程。

参见代码

/**
     *类说明：自定义线程池中线程的创建方式，把线程设置为守护线程
     */
    public class ThreadPoolAdv {
        static class Worker implements Runnable
        {
            private String taskName;
            private Random r = new Random();
    
            public Worker(String taskName){
                this.taskName = taskName;
            }
    
            public String getName() {
                return taskName;
            }
    
            @Override
            public void run(){
                System.out.println(Thread.currentThread().getName()
                		+" process the task : " + taskName);
                SleepTools.ms(r.nextInt(100)*5);
            }
        }
    
        private static class MyThreadFactory implements ThreadFactory{
    
            private AtomicInteger count = new AtomicInteger(0);
    
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r,"nandao_"+count.getAndIncrement());
                t.setDaemon(true);
                System.out.println("create "+t);
                return t;
            }
        }
    
        public static void main(String[] args) throws InterruptedException, ExecutionException
        {
            ExecutorService threadPool = new ThreadPoolExecutor(2,
                    4, 3,
                    TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10),
                    new MyThreadFactory(),
                    new ThreadPoolExecutor.DiscardOldestPolicy());
    
            for (int i = 0; i <= 6; i++)
            {
                Worker worker = new Worker("worker " + i);
                System.out.println("A new task has been added : " + worker.getName());
                 // threadPool.execute(worker);
                //打印每个线程的名称，常用功能
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"OK");
                });
                Thread.sleep(100);
            }
    
        }
    }

Executors  静态工厂里默认的  threadFactory ，线程的命名规则是“ pool- 数字 -thread-数字”。

**RejectedExecutionHandler**

线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提 交任务，必须采取一种策略处理该任务，线程池提供了 4  种策略：

（1 ） AbortPolicy ：直接抛出异常，默认策略；

（ 2 ） CallerRunsPolicy ：用调用者所在的线程来执行任务；

（ 3 ） DiscardOldestPolicy ：丢弃阻塞队列中靠最前的任务，并执行当前任务；

（ 4 ） DiscardPolicy ：直接丢弃任务；

当然也可以根据应用场景实现  RejectedExecutionHandler  接口，自定义饱和 策略，如记录日志或持久化存储不能处理的任务，后续单独处理。

**扩展线程池**

能扩展线程池的功能吗？比如在任务执行的前后做一点我们自己的业务工 作？实际上，JDK 的线程池已经为我们预留的接口，在线程池核心方法中，有 2 个方法是空的，就是给我们预留的。还有一个线程池退出时会调用的方法。

测试代码：

/**
     *类说明：扩展线程池的使用范例
     */
    public class ThreadPoolExt {
        static class Worker implements Runnable
        {
            private String taskName;
            private Random r = new Random();
    
            public Worker(String taskName){
                this.taskName = taskName;
            }
    
            public String getName() {
                return taskName;
            }
    
            @Override
            public void run(){
                //打印线程名称方便定位bug
                System.out.println(Thread.currentThread().getName()
                		+" process the task : " + taskName);
                SleepTools.ms(r.nextInt(100)*5);
            }
        }
    
    
        public static void main(String[] args)
                throws InterruptedException, ExecutionException
        {
            ExecutorService threadPool = new ThreadPoolExecutor(2, 4, 3,
                    TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10),
                    new ThreadPoolExecutor.DiscardOldestPolicy()){
                @Override
                protected void beforeExecute(Thread t, Runnable r) {
                    System.out.println("Ready Execute "+((Worker)r).getName());
                }
    
                @Override
                protected void afterExecute(Runnable r, Throwable t) {
                    System.out.println("Complete Execute "+((Worker)r).getName());
                }
    
                @Override
                protected void terminated() {
                    System.out.println("线程池退出 ");
                }
            };
    
            for (int i = 0; i <= 3; i++)
            {
                Worker worker = new Worker("worker " + i);
                System.out.println("A new task has been added : " + worker.getName());
                threadPool.execute(worker);
            }
            threadPool.shutdown();
    
        }
    }

可以看到，每个任务执行前后都会调用 beforeExecute 和 afterExecute 方法。 相当于执行了一个切面。而在调用 shutdown 方法后则会调用 terminated 方法。

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

**线程池的工作机制**

1 ）如果当前运行的线程少于  corePoolSize ，则创建新线程来执行任务（注意， 执行这一步骤需要获取全局锁）。

2 ）如果运行的线程等于或多于  corePoolSize ，则将任务加入  BlockingQueue 。

3 ）如果无法将任务加入  BlockingQueue （队列已满），则创建新的线程来处 理任务。

4 ）如果创建新线程将使当前运行的线程超出  maximumPoolSize ，任务将被 拒绝，并调用 RejectedExecutionHandler.rejectedExecution() 方法。

**提交任务**

execute() 方法用于提交不需要返回值的任务，所以无法判断任务是否被线程 池执行成功。

submit() 方法用于提交需要返回值的任务。线程池会返回一个  future  类型的 对象，通过这个 future  对象可以判断任务是否执行成功，并且可以通过  future 的 get() 方法来获取返回值， get() 方法会阻塞当前线程直到任务完成，而使用  get （long timeout ， TimeUnit unit ）方法则会阻塞当前线程一段时间后立即返回，这 时候有可能任务没有执行完。

**关闭线程池**

可以通过调用线程池的  shutdown  或  shutdownNow  方法来关闭线程池。它们 的原理是遍历线程池中的工作线程，然后逐个调用线程的 interrupt  方法来中断 线程，所以无法响应中断的任务可能永远无法终止。但是它们存在一定的区别， shutdownNow 首先将线程池的状态设置成  STOP ，然后尝试停止所有的正在执行 或暂停任务的线程，并返回等待执行任务的列表，而 shutdown  只是将线程池的 状态设置成 SHUTDOWN  状态，然后中断 **所有没有正在执行任务的线程** 。

只要调用了这两个关闭方法中的任意一个，isShutdown 方法就会返回  true 。 当所有的任务都已关闭后，才表示线程池关闭成功，这时调用 isTerminaed  方法 会返回 true 。至于应该调用哪一种方法来关闭线程池，应该由提交到线程池的任 务特性决定，通常调用 shutdown  方法来关闭线程池，如果任务不一定要执行完， 则可以调用 shutdownNow  方法。

**合理地配置线程池**

要想合理地配置线程池，就必须首先分析任务特性，可以从以下几个角度来 分析。

•任务的性质： CPU  密集型任务、 IO  密集型任务和混合型任务。

•任务的优先级：高、中和低。

•任务的执行时间：长、中和短。

•任务的依赖性：是否依赖其他系统资源，如数据库连接。 性质不同的任务可以用不同规模的线程池分开处理。

CPU 密集型任务应配置尽可能小的线程，如配置  Ncpu+1  个线程的线程池。什么是cup密集型？

在一个任务中，主要做计算，CPU持续在运行，CPU利用率高，具有这种特点的任务称为CPU密集型任务。

线程个数为CPU+1核数。这几个线程可以并行执行，不存在线程切换到开销，提高了cpu的利用率的同时也减少了切换线程导致的性能损耗，+1的目的是考虑到缓存页的情况，如果出现这种情况，则多占一个线程。

由于 IO  密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如 2\*Ncpu。什么是io密集型？

在一个任务中，大部分时间在进行I/O操作，由于I/O速度远远小于CPU，所以任务的大部分时间都在等待IO，CPU利用率低，具有这种特点的任务称为IO密集型任务

线程个数为CPU核数的两倍的原因是线程在IO操作的时候，其他线程可以继续用cpu，提高了cpu的利用率。

混合型的任务，如果可以拆分，将其拆分成一个  CPU  密集型任务和一个  IO 密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐 量将高于串行执行的吞吐量。如果这两个任务执行时间相差太大，则没必要进行 分解。可以通过 Runtime.getRuntime().availableProcessors() 方法获得当前设备的 CPU 个数。 对于 IO  型的任务的最佳线程数，有个公式可以计算 Nthreads = NCPU \* UCPU \* (1 + W/C)

其中：

❑ NCPU  是处理器的核的数目

❑ UCPU  是期望的  CPU  利用率（该值应该介于  0  和  1  之间）

❑ W/C  是等待时间与计算时间的比率

等待时间与计算时间我们在 Linux  下使用相关的  vmstat  命令或者  top  命令查 看。优先级不同的任务可以使用优先级队列 PriorityBlockingQueue  来处理。它可 以让优先级高的任务先执行。执行时间不同的任务可以交给不同规模的线程池来处理，或者可以使用优先 级队列，让执行时间短的任务先执行。 依赖数据库连接池的任务，因为线程提交 SQL 后需要等待数据库返回结果， 等待的时间越长，则 CPU 空闲时间就越长，那么线程数应该设置得越大，这样 才能更好地利用 CPU。

建议使用有界队列。有界队列能增加系统的稳定性和预警能力，可以根据需 要设大一点儿，比如几千。 假设，我们现在有一个 Web 系统，里面使用了线程池来处理业务，在某些

情况下，系统里后台任务线程池的队列和线程池全满了，不断抛出抛弃任务的异 常，通过排查发现是数据库出现了问题，导致执行 SQL  变得非常缓慢，因为后台 任务线程池里的任务全是需要向数据库查询和插入数据的，所以导致线程池里的 工作线程全部阻塞，任务积压在线程池里。 如果当时我们设置成无界队列，那么线程池的队列就会越来越多，有可能会 撑满内存，导致整个系统不可用，而不只是后台任务出现问题。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/5598992acb3e4c298646105140224c6a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221120/acd880d83ff04df09b1890079426d9dd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221120/c1b2b3eb4fb944699bc595dbdf51e843.png