线程池的使用、工作原理和优势

分手后的思念是犯贱 2023-10-14 08:37 7阅读 0赞

> 关于作者：CSDN内容合伙人、技术专家， 从零开始做日活千万级APP。  
> 专注于分享各领域原创系列文章 ，擅长java后端、移动开发、人工智能等，希望大家多多支持。

#### 目录 ####

*  一、导读
 *  二、线程池概览
 *   *  2.1 为什么创建和销毁线程开销较大
     *  2.2 为什么要使用线程池？
     *  2.3 在配置线程池的时候需要考虑哪些配置因素?
 *  三、线程池使用
 *   *  3.1 线程池的创建
     *   *  3.1.1 newFixedThreadPool
         *  3.1.2 newCachedThreadPool
         *  3.1.3 newScheduledThreadPool
         *  3.1.4 newSingleThreadExecutor
     *  3.2 ThreadPoolExecutor
     *  3.3 为什么线程池不推荐使用Executors去创建，而是通过ThreadPoolExecutor的方式
 *  四、线程池原理
 *   *  4.1 线程池中常用的workQueue（缓冲队列）
     *  4.2 线程池中拒绝策略
     *  4.3 线程池任务的关闭
     *  4.4 线程的复用
 *  五、 推荐阅读

![ddd][]

## 一、导读 ##

我们继续总结学习**Java基础知识**，温故知新。

线程池。

## 二、线程池概览 ##

在Java中，创建和销毁线程开销较大，为了避免线程过多而带来使用上的开销。  
所以我们需要对线程进行统一管理及复用，这就是我们要说的线程池。

线程池用于管理和复用多个线程，把一个或多个线程通过统一的方式进行调度和重复使用的技术。

从JDK 5开始，把工作单元与执行机制分离开来，工作单元包括Runnable和Callable，而执行机制由Executor框架提供。

### 2.1 为什么创建和销毁线程开销较大 ###

创建和销毁线程的开销较大主要是因为涉及到以下几个方面：

1.  上下文切换：在多线程环境中，当一个线程被创建或销毁时，操作系统需要切换上下文，将CPU的执行权从一个线程转移到另一个线程。这个过程涉及保存和恢复线程的状态信息，包括寄存器值、栈指针和程序计数器等。上下文切换是一项耗时的操作，会导致额外的开销。
2.  内存管理：每个线程需要分配一定的内存空间来存储线程的堆栈、线程私有数据等。创建和销毁线程会涉及内存的分配和释放，而内存分配和释放操作通常比较耗时。
3.  调度开销：操作系统需要进行调度，决定哪个线程应该获得CPU的执行权。线程的创建和销毁会引起调度器的重新调度，这涉及到时间片、优先级和调度算法等方面的开销。
4.  同步和通信：多线程编程中，线程之间通常需要进行同步和通信，以确保数据的一致性和线程间的协调。创建和销毁线程会涉及到锁、信号量、管道等同步和通信机制的初始化和清理，增加了开销。

### 2.2 为什么要使用线程池？ ###

1.  降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
2.  提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
3.  提高线程的可管理性。

### 2.3 在配置线程池的时候需要考虑哪些配置因素? ###

从任务的优先级，任务的执行时间长短，任务的性质(CPU密集/ IO密集)，任务的依赖关系这四个角度来分析。并且近可能地使用有界的工作队列。  
性质不同的任务可用使用不同规模的线程池分开处理:

*  CPU密集型: 尽可能少的线程，Ncpu+1
 *  IO密集型: 尽可能多的线程, Ncpu\*2，比如数据库连接池
 *  混合型: CPU密集型的任务与IO密集型任务的执行时间差别较小，拆分为两个线程池；否则没有必要拆分

[CPU密集型和IO密集型任务的权衡：如何找到最佳平衡点][CPU_IO]

## 三、线程池使用 ##

### 3.1 线程池的创建 ###

java提供了多种方式：

1. newFixedThreadPool(n):创建一个数量固定的线程池，超出的任务会在队列中等待空闲的线程，
    可用于控制程序的最大并发数。
     
    2. newCacheThreadPool():短时间内处理大量工作的线程池，会根据任务数量产生对应的线程，
    并试图缓存线程以便重复使用，如果限制 60 秒没被使用，则会被移除缓存。如果现有线程没有可用的，
    则创建一个新线程并添加到池中，如果有被使用完但是还没销毁的线程，就复用该线程。
    终止并从缓存中移除那些已有 60 秒钟未被使用的线程。因此，长时间保持空闲的线程池不会使用任何资源。
     
    3. newScheduledThreadPool()：创建一个数量固定的线程池，支持执行定时性或周期性任务。
     
    4. newWorkStealingPool(n)：Java 8 新增创建线程池的方法，创建时如果不设置任何参数，
    则以当前机器CPU 处理器数作为线程个数，此线程池会并行处理任务，不能保证执行顺序。
     
    5. newSingleThreadExecutor()：创建一个单线程的线程池。这个线程池只有一个线程在工作，
    也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。
    此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
     
    6. newSingleThreadScheduledExecutor()：此线程池就是单线程的 newScheduledThreadPool。

#### 3.1.1 newFixedThreadPool ####

创建固定大小的线程池，比如线程池容量是10，最多可以同时执行10个线程。  
使用案例

创建线程池,参数是创造的线程数量
    ExecutorService pool = Executors.newFixedThreadPool(10);
    for (int i = 0; i < 10; i++) {
        
       int j = i;
       pool.execute(new Runnable() {
        
           @Override
           public void run() {
        
           }
       });
    }

#### 3.1.2 newCachedThreadPool ####

创建一个可缓存的线程池，此线程池**不会对线程池大小做限制**，线程池大小完全依赖于JVM能够创建的最大线程大小，当然线程池里的线程是可以复用的，但是如果在高并发的情况下，这个线程池在会导致运行时内存溢出问题。  
使用案例

ExecutorService executorService = Executors.newCachedThreadPool();
    for (int i = 0; i < 6; i++) {
        
       int j = i;
       executorService.execute(new Runnable() {
        
           @Override
           public void run() {
        
           }
       });
    }

#### 3.1.3 newScheduledThreadPool ####

创建一个**定时**执行的线程池，里边提供了两个方法，FixRate和fixDelay，  
fixRate 就是以固定时间周期执行任务，不管上一个线程是否执行完，  
fixDelay 的话就是以固定的延迟执行任务，就是在上一个任务执行完成之后，延迟一定时间执行。  
使用案例

ScheduledExecutorService executorService = Executors.newScheduledThreadPool(2);
    for (int i = 0; i < 5; i++) {
        
       int j = i;
       executorService.schedule(new Runnable() {
        
           @Override
           public void run() {
        
           }
       }, 3L, TimeUnit.SECONDS);
    }

#### 3.1.4 newSingleThreadExecutor ####

创建一个单线程的线程池，这个线程池同时只能执行一个线程，可以保证线程按顺序执行，保证数据安全。  
使用案例

public class SingleThreadPoolDemo {
        
        //格式化
        static SimpleDateFormat sim = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //AtomicInteger用来计数
        static AtomicInteger number = new AtomicInteger();
        public static void main(String[] args) {
        
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            for (int i = 0; i < 6; i++) {
        
                executorService.execute(new Runnable() {
        
                    @Override
                    public void run() {
        
                    }
                });
            }
        }
    }

### 3.2 ThreadPoolExecutor ###

通过ThreadPoolExecutor的方式创建线程池，前面四种都不是我们推荐都方式

public class ThreadPoolDemo {
        
        private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
          2,   // 核心线程数
          10,  // 最大线程数
          10L, // 线程存活时间
          TimeUnit.SECONDS,  // 线程存活时间单位
          new LinkedBlockingQueue(100));// 缓冲队列
        
        public static void main(String[] args) {
        
            threadPoolExecutor.execute(new Runnable() {
        
                @Override
                public void run() {
        
                }
            });
        }
    }

### 3.3 为什么线程池不推荐使用Executors去创建，而是通过ThreadPoolExecutor的方式 ###

newFixedThreadPool（固定线程数）  
newSingleThreadExecutor（单线程）

*  主要问题是堆积的请求处理队列可能会耗费非常大的内存，甚至OOM。线程数固定，任务多了之后容易堆积。

newCachedThreadPool（可缓存的线程池）  
newScheduledThreadPool（定时执行的线程池）

*  主要问题是线程数最大数是Integer.MAX\_VALUE，可能会创建数量非常多的线程，甚至OOM。不限定线程多数量，任务一多，容易创建无限多线程。

## 四、线程池原理 ##

先看个图，方便理解  
![在这里插入图片描述][d9a62c49bf034f359aed533b809d75f2.png]

1.  当有新的任务进来时，线程池将**当前线程数量**与**核心数量**进行比较，如果没有超过核心数就会新建线程进行任务执行，
2.  如果已经超过核心线程数，则判断**缓冲队列**是否已经满了，没有满的话任务就会被放入缓冲队列中排队等待执行；
3.  如果缓冲队列超过最大队列数，并且线程池没有达到**最大线程数**，就会新建线程来执行任务；
4.  如果超过了最大线程数，就会执行拒绝执行策略。

再简单点，就两个队列，一个线程集合workerSet和一个阻塞队列workQueue  
![在这里插入图片描述][acb94aed32234edb83f36f61ec086868.png]

![在这里插入图片描述][c2d0269a79ab4cab91340d4bbaccf432.png]

我们一起来看看源码，  
线程池类为 java.util.concurrent.ThreadPoolExecutor，常用构造方法为：

ThreadPoolExecutor(
       int corePoolSize,   // 核心线程数
       int maximumPoolSize,   // 最大线程数
       long keepAliveTime,     // 线程存活时间
       TimeUnit unit,          // 线程存活时间单位
       BlockingQueue<Runnable> workQueue,  // 缓冲队列
       RejectedExecutionHandler handler    // 拒绝策略
    )

处理任务的优先级为：核心线程 > 缓冲队列 > 最大线程  
corePoolSize > workQueue > maximumPoolSize

如果三者都满了，使用handler处理被拒绝的任务。

/**
     * 将该Runnable任务加入线程池并在未来某个时刻执行
     * 该任务可能执行在一个新的线程 或 一个已存在的线程池中的线程
     * 如果该任务提交失败,可能是因为线程池已关闭,或者已达到线程池队列和线程数已满.
     * 该Runnable将交给RejectedExecutionHandler处理,抛出RejectedExecutionException
     */
    public void execute(Runnable command) {
        
        if (command == null){
        
            //如果没传入Runnable任务,则抛出空指针异常
            throw new NullPointerException();
        }
        
        int c = ctl.get();
        //当前线程数 小于 核心线程数
        if (workerCountOf(c) < corePoolSize) {
        
            //直接开启新的线程,并将Runnable传入作为第一个要执行的任务,成功返回true,否则返回false
            if (addWorker(command, true)){
        
                return;
            }
            c = ctl.get();
        }
    
        //c < SHUTDOWN代表线程池处于RUNNING状态 + 将Runnable添加到任务队列,如果添加成功返回true失败返回false
        if (isRunning(c) && workQueue.offer(command)) {
        
            int recheck = ctl.get();
            //成功加入队列后,再次检查是否需要添加新线程(因为已存在的线程可能在上次检查后销毁了,或者线程池在进入本方法后关闭了)
            if (! isRunning(recheck) && remove(command)){
        
                //如果线程池处于非RUNNING状态 并且 将该Runnable从任务队列中移除成功,则拒绝执行此任务
                //交给RejectedExecutionHandler调用rejectedExecution方法,拒绝执行此任务
                reject(command);
            }else if (workerCountOf(recheck) == 0){
        
                //如果线程池线程数量为0,则创建一条新线程,去执行
                addWorker(null, false);
            }   
        }else if (!addWorker(command, false))
            //如果线程池处于非RUNNING状态 或 将Runnable添加到队列失败(队列已满导致),则执行默认的拒绝策略
            reject(command);
    }

当ThreadPoolExecutor创建新线程时，通过CAS来更新线程池的状态ctl.

### 4.1 线程池中常用的workQueue（缓冲队列） ###

1.  ArrayBlockingQueue(有界缓存等待队列)  
    可以指定缓存队列的大小
2.  LinkedBlockingQueue(无界缓存等待队列)  
    可以创建Integer.MAX\_VALUE个线程，容易OOM

当前执行的线程数量达到corePoolSize（核心）的数量时，剩余的元素会在阻塞队列里等待，在使用此阻塞队列时maximumPoolSizes就相当于无效了。

1.  SynchronousQueue(无缓冲等待队列)  
    是一个不存储元素的阻塞队列，会直接将任务交给消费者，必须等队列中的添加元素被消费后才能继续添加新的元素。使用SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界，避免线程拒绝执行操作

### 4.2 线程池中拒绝策略 ###

所有拒绝策略都实现了接口 RejectedExecutionHandler

public interface RejectedExecutionHandler {
        
        /**
         * @param r  待执行任务
         * @param executor 线程池
         * @throws RejectedExecutionException  方法可能会抛出拒绝异常
         */
        void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
    }

1.AbortPolicy  
直接抛出拒绝异常，会中断调用者的处理过程，所以除非有明确需求，一般不推荐

public static class AbortPolicy implements RejectedExecutionHandler {
        
         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        
             throw new RejectedExecutionException("Task " + r.toString() +
                                                  " rejected from " +
                                                  e.toString());
         }
     }

2.CallerRunsPolicy  
在调用者线程中运行当前被丢弃的任务,也就是说谁把 runnable 这个任务甩出来。  
用调用者所在线程来运行任务，也就是说任务不会进入线程池。  
**如果线程池已经被关闭，则直接丢弃该任务**。

public static class CallerRunsPolicy implements RejectedExecutionHandler {
        
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        
            if (!e.isShutdown()) {
        
                r.run();
            }
        }
    }

1.  DiscardOledestPolicy  
    丢弃队列中最老的，然后再次尝试提交新任务。

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
        
         public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        
             if (!e.isShutdown()) {
        
                 //获得待执行的任务队列,队列先进先出
                 //poll()方法就能直接把队列中最老的抛弃掉，再次尝试执行execute(r)
                 e.getQueue().poll();
                 e.execute(r);
             }
         }
     }

1.  DiscardPolicy  
    默默丢弃无法加载的任务。这个代码就很简单了，真的是啥也没做。

public static class DiscardPolicy implements RejectedExecutionHandler {
        
      public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        
      }
    }

1.  自定义拒绝策略  
    只要继承接口都可以根据自己需要自定义拒绝策略.

**案例1：**  
单独启动一个新的临时线程来执行任务。

private class NewThreadRunsPolicy implements RejectedExecutionHandler {
        
      public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        
          try {
        
              final Thread t = new Thread(r, "Temporary task executor");
              t.start();
          } catch (Throwable e) {
        
              throw new RejectedExecutionException(
                      "Failed to start a new thread", e);
          }
      }
    }

**案例2：**  
直接继承的 AbortPolicy ，加强了日志输出，并且输出dump文件,然后任务也是拒绝

public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
        
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        
            String msg = String.format("Thread pool is EXHAUSTED!" +
                            " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
                            " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
                    threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
                    e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
                    url.getProtocol(), url.getIp(), url.getPort());
            logger.warn(msg);
            dumpJStack();
            throw new RejectedExecutionException(msg);
        }
    }

### 4.3 线程池任务的关闭 ###

shutdown方法会将线程池的状态设置为SHUTDOWN,线程池进入这个状态后,就拒绝再接受任务,然后会将剩余的任务全部执行完

1.  临时线程什么时候创建？  
    新任务提交时发现核心线程都在忙，任务队列也满了，并且还可以创建临时线程，  
    此时才会创建临时线程；
2.  什么时候会开始拒绝任务？  
    核心线程和临时线程都在忙，任务队列也满了，新的任务过来的时候才会开始拒绝任务

### 4.4 线程的复用 ###

在ThreadPoolExecutor.java的runwork方法中通过一个while循环,不断的getTask()取任务出来执行,以这种方式实现了线程的复用.

## 五、 推荐阅读 ##

[Java 专栏][Java]

[SQL 专栏][SQL]

[数据结构与算法][Link 1]

[Android学习专栏][Android]

![在这里插入图片描述][f5ba2a630958460ab6a581523327cf4d.png]

[ddd]: https://img-blog.csdnimg.cn/004fcf994a0b443ab491ca3bd590e764.png
[CPU_IO]: http://t.csdn.cn/c7I5d
[d9a62c49bf034f359aed533b809d75f2.png]: https://img-blog.csdnimg.cn/d9a62c49bf034f359aed533b809d75f2.png
[acb94aed32234edb83f36f61ec086868.png]: https://img-blog.csdnimg.cn/acb94aed32234edb83f36f61ec086868.png
[c2d0269a79ab4cab91340d4bbaccf432.png]: https://img-blog.csdnimg.cn/c2d0269a79ab4cab91340d4bbaccf432.png
[Java]: https://blog.csdn.net/fumeidonga/category_12311533.html
[SQL]: http://t.csdn.cn/4i3TW
[Link 1]: http://t.csdn.cn/9gGNy
[Android]: http://t.csdn.cn/A0aTR
[f5ba2a630958460ab6a581523327cf4d.png]: https://img-blog.csdnimg.cn/f5ba2a630958460ab6a581523327cf4d.png