线程池的内部结构与原理解析

偏执的太偏执、 2024-03-30 16:04 96阅读 0赞

## 线程池的内部结构 ##

*  使⽤线程池的原因：
    
    > 1、降低资源消耗
    
    > 2、控制并发的数量。并发数量过多，可能会导致资源消耗过多，从⽽造成服务器崩溃
    
    > 3、提高线程的可管理性
    
    > > 1、线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。
    
    > 4、提高响应速度
    > 
    > > 1、当任务到达时，任务可以不需要等到线程创建就能立即执行。

### 1、线程池的结构 ###

> 1、线程池的内部结构主要由四部分组成
     >> 1、第一部分是线程池管理器，它主要负责管理线程池的创建、销毁、添加任务等管理操作，它是整个线程池的管家。
     >> 2、第二部分是工作线程，也就是图中的线程 t0~t9，这些线程勤勤恳恳地从任务队列中获取任务并执行。
     >> 3、第三部分是任务队列，作为一种缓冲机制，线程池会把当下没有处理的任务放入任务队列中，由于多线程同时从任务队列中获取任务是并发场景，此时就需要任务队列满足线程安全的要求，所以线程池中任务队列采用 BlockingQueue 来保障线程安全。
    >> 4、 第四部分是任务，任务要求实现统一的接口，以便工作线程可以处理和执行。

![线程池结构][adbdda1a828740fcbd7e93ffc737456f.png]

> 2、阻塞队列  
> ![在这里插入图片描述][f724be57e94c475b843e9d2a0d8f5bc4.png]

> 3、线程池中的这四个主要组成部分最值得我们关注的就是阻塞队列了，如表格所示，不同的线程池会选用不同的阻塞队列。

*  **LinkedBlockingQueue**

> 1、对于 FixedThreadPool 和 SingleThreadExector 而言，它们使用的阻塞队列是容量为 Integer.MAX\_VALUE 的 LinkedBlockingQueue，可以认为是无界队列。由于 FixedThreadPool 线程池的线程数是固定的，所以没有办法增加特别多的线程来处理任务，这时就需要 LinkedBlockingQueue 这样一个没有容量限制的阻塞队列来存放任务。**这里需要注意，由于线程池的任务队列永远不会放满，所以线程池只会创建核心线程数量的线程，所以此时的最大线程数对线程池来说没有意义，因为并不会触发生成多于核心线程数的线程。**

*  **SynchronousQueue**

> 1、第二种阻塞队列是 SynchronousQueue，对应的线程池是 CachedThreadPool。线程池 CachedThreadPool 的最大线程数是 Integer 的最大值，可以理解为线程数是可以无限扩展的。CachedThreadPool 和上一种线程池 FixedThreadPool 的情况恰恰相反，**FixedThreadPool 的情况是阻塞队列的容量是无限的，而这里 CachedThreadPool 是线程数可以无限扩展，所以 CachedThreadPool 线程池并不需要一个任务队列来存储任务，因为一旦有任务被提交就直接转发给线程或者创建新线程来执行，而不需要另外保存它们。**

*  **DelayedWorkQueue**
    
    > 1、第三种阻塞队列是DelayedWorkQueue，它对应的线程池分别是 ScheduledThreadPool 和 SingleThreadScheduledExecutor，这两种线程池的最大特点就是可以延迟执行任务，比如说一定时间后执行任务或是每隔一定的时间执行一次任务。DelayedWorkQueue 的特点是内部元素并不是按照放入的时间排序，而是会按照延迟的时间长短对任务进行排序，内部采用的是“堆”的数据结构。之所以线程池 ScheduledThreadPool 和 SingleThreadScheduledExecutor 选择 DelayedWorkQueue，是因为它们本身正是基于时间执行任务的，而延迟队列正好可以把任务按时间进行排序，方便任务的执行

### 2、线程池的原理 ###

> 1、线程池的基本结构，如下图所示。用户通过使用线程池的execute方法将Runnable提交到线程池中进行执行。当线程池中的线程都很忙的时候，这个新加入的 Runnable 就会被放到等待队列。当线程空闲下来的时候，就会去等待队列里查看是否还有排队等待的任务，如果有，就会队列中取出任务并继续执行。如果没有，线程就会进入休眠。  
> ![在这里插入图片描述][2091e681eabc4a4ea4fc721be77a61a9.png]

> 2、如果我们更具体一点，就以本周第二节课所讲的ThreadPoolExecutor为例。当我们把一个Runnable交给线程池去执行的时候，这个线程池处理的流程是这样的：
> 
> > 2.1、先判断线程池中的核心线程们是否空闲，如果空闲，就把这个新的任务指派给某一个空闲线程去执行。如果没有空闲，并且当前线程池中的核心线程数还小于 corePoolSize，那就再创建一个核心线程。  
> > 2.2、果线程池的线程数已经达到核心线程数，并且这些线程都繁忙，就把这个新来的任务放到等待队列中去。如果等待队列又满了，那么查看一下当前线程数是否到达maximumPoolSize，如果还未到达，就继续创建线程。如果已经到达了，就交给RejectedExecutionHandler来决定怎么处理这个任务。

public void execute(Runnable command) {
        
            if (command == null)
                throw new NullPointerException();
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) {
        
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            if (isRunning(c) && workQueue.offer(command)) {
        
                int recheck = ctl.get();
                if (! isRunning(recheck) && remove(command))
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            else if (!addWorker(command, false))
                reject(command);
        }

*  Java中的线程池是用ThreadPoolExecutor类来实现的,先看一下线程池的类图：  
    ![20220811.png][]

#### 2.1、线程池的创建 ####

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
     
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }
     
    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 ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

参数介绍
    创建线程池需要以下几个参数，其中有5个是必需的：
    
    1、corePoolSize（必需）：核心线程数。默认情况下，核心线程会一直存活，但是当将 allowCoreThreadTimeout 设置为 true 时，核心线程也会超时回收。
    2、maximumPoolSize（必需）：线程池所能容纳的最大线程数。当活跃线程数达到该数值后，后续的新任务将会阻塞。
    3、keepAliveTime（必需）：线程闲置超时时长。如果超过该时长，非核心线程就会被回收。如果将 allowCoreThreadTimeout 设置为 true 时，核心线程也会超时回收。
    4、unit（必需）：指定 keepAliveTime 参数的时间单位。常用的有：TimeUnit.MILLISECONDS（毫秒）、TimeUnit.SECONDS（秒）、TimeUnit.MINUTES（分）。
    5、workQueue（必需）：任务队列。通过线程池的 execute() 方法提交的 Runnable 对象将存储在该参数中。其采用阻塞队列实现。
    6、threadFactory（可选）：线程工厂。用于指定为线程池创建新线程的方式。
    7、handler（可选）：拒绝策略。当达到最大线程数时需要执行的饱和策略。

#### 2.2、使用线程池 ####

// 创建线程池
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE, TimeUnit.SECONDS, sPoolWorkQueue, sThreadFactory);
    // 向线程池提交任务
    threadPool.execute(new Runnable() {
        
     @Override
     public void run() {
        
         ... // 线程执行的任务
     }
    });
    // 关闭线程池
    threadPool.shutdown(); // 设置线程池的状态为SHUTDOWN，然后中断所有没有正在执行任务的线程
    threadPool.shutdownNow(); // 设置线程池的状态为 STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表

#### 2.3、线程池原理 ####

*  下面来描述一下线程池工作的原理，同时对上面的参数有一个更深的了解。其工作原理流程图如下：  
    ![459d0f554d5d4b5a34157f07e60cd27d.png][]

#### 2.4、线程池的参数 ####

##### 2.4.1、任务队列（workQueue） #####

*  任务队列是基于阻塞队列实现的，即采用生产者消费者模式，在 Java 中需要实现 BlockingQueue 接口。但 Java 已经为我们提供了 7 种阻塞队列的实现：

> 1、ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列（数组结构可配合指针实现一个环形队列）。  
> LinkedBlockingQueue： 一个由链表结构组成的有界阻塞队列，在未指明容量时，容量默认为 Integer.MAX\_VALUE。

> 2、PriorityBlockingQueue： 一个支持优先级排序的无界阻塞队列，对元素没有要求，可以实现 Comparable 接口也可以提供 Comparator 来对队列中的元素进行比较。跟时间没有任何关系，仅仅是按照优先级取任务。

> 3、DelayQueue：类似于PriorityBlockingQueue，是二叉堆实现的无界优先级阻塞队列。要求元素都实现 Delayed 接口，通过执行时延从队列中提取任务，时间没到任务取不出来。

> 4、SynchronousQueue： 一个不存储元素的阻塞队列，消费者线程调用 take() 方法的时候就会发生阻塞，直到有一个生产者线程生产了一个元素，消费者线程就可以拿到这个元素并返回；生产者线程调用 put() 方法的时候也会发生阻塞，直到有一个消费者线程消费了一个元素，生产者才会返回。

> 5、LinkedBlockingDeque： 使用双向队列实现的有界双端阻塞队列。双端意味着可以像普通队列一样 FIFO（先进先出），也可以像栈一样 FILO（先进后出）。

> 6、LinkedTransferQueue： 它是ConcurrentLinkedQueue、LinkedBlockingQueue 和 SynchronousQueue 的结合体，但是把它用在 ThreadPoolExecutor 中，和 LinkedBlockingQueue 行为一致，但是是无界的阻塞队列。

*  *注意有界队列和无界队列的区别：如果使用有界队列，当队列饱和时并超过最大线程数时就会执行拒绝策略；而如果使用无界队列，因为任务队列永远都可以添加任务，所以设置maximumPoolSize 没有任何意义*

#### 2.5、线程工厂（threadFactory） ####

> 1、线程工厂指定创建线程的方式，需要实现 ThreadFactory 接口，并实现 newThread(Runnable r) 方法。该参数可以不用指定，Executors 框架已经为我们实现了一个默认的线程工厂：

/**
    * The default thread factory.
    */
    private static class DefaultThreadFactory implements ThreadFactory {
        
     private static final AtomicInteger poolNumber = new AtomicInteger(1);
     private final ThreadGroup group;
     private final AtomicInteger threadNumber = new AtomicInteger(1);
     private final String namePrefix;
    
     DefaultThreadFactory() {
        
         SecurityManager s = System.getSecurityManager();
         group = (s != null) ? s.getThreadGroup() :
                               Thread.currentThread().getThreadGroup();
         namePrefix = "pool-" +
                       poolNumber.getAndIncrement() +
                      "-thread-";
     }
    
     public Thread newThread(Runnable r) {
        
         Thread t = new Thread(group, r,
                               namePrefix + threadNumber.getAndIncrement(),
                               0);
         if (t.isDaemon())
             t.setDaemon(false);
         if (t.getPriority() != Thread.NORM_PRIORITY)
             t.setPriority(Thread.NORM_PRIORITY);
         return t;
     }
    }

#### 2.6、拒绝策略（handler） ####

*  当线程池的线程数达到最大线程数时，需要执行拒绝策略。拒绝策略需要实现 RejectedExecutionHandler 接口，并实现 rejectedExecution(Runnable r, ThreadPoolExecutor executor) 方法。不过 Executors 框架已经为我们实现了 4 种拒绝策略：

> 1、AbortPolicy（默认）：丢弃任务并抛出RejectedExecutionException 异常。  
> 2、CallerRunsPolicy：由调用线程处理该任务。  
> 3、DiscardPolicy：丢弃任务，但是不抛出异常。可以配合这种模式进行自定义的处理方式。  
> 4、DiscardOldestPolicy：丢弃队列最早的未处理任务，然后重新尝试执行任务。

### 2.5、功能线程池 ###

*  定长线程池（FixedThreadPool）
 *  定时线程池（ScheduledThreadPool ）
 *  可缓存线程池（CachedThreadPool）
 *  单线程化线程池（SingleThreadExecutor）

#### 2.5.1、定长线程池（FixedThreadPool） ####

*  源码

public static ExecutorService newFixedThreadPool(int nThreads) {
        
      return new ThreadPoolExecutor(nThreads, nThreads,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>());
    }
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        
      return new ThreadPoolExecutor(nThreads, nThreads,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory);
    }
    
    
    特点：只有核心线程，线程数量固定，执行完立即回收，任务队列为链表结构的有界队列。
    应用场景：控制线程最大并发数

*  使用示例：

// 1. 创建定长线程池对象 & 设置线程池线程数量固定为3
    ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
    // 2. 创建好Runnable类线程对象 & 需执行的任务
    Runnable task =new Runnable(){
        
      public void run() {
        
         System.out.println("执行任务啦");
      }
    };
    // 3. 向线程池提交任务
    fixedThreadPool.execute(task);

#### 2.5.2、定时线程池（ScheduledThreadPool ） ####

*  源码

private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;
    
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        
       return new ScheduledThreadPoolExecutor(corePoolSize);
    }
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        
       super(corePoolSize, Integer.MAX_VALUE,
             DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
             new DelayedWorkQueue());
    }
    
    public static ScheduledExecutorService newScheduledThreadPool(
           int corePoolSize, ThreadFactory threadFactory) {
        
       return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
    }
    public ScheduledThreadPoolExecutor(int corePoolSize,
                                      ThreadFactory threadFactory) {
        
       super(corePoolSize, Integer.MAX_VALUE,
             DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
             new DelayedWorkQueue(), threadFactory);
    }
    
    
    特点：核心线程数量固定，非核心线程数量无限，执行完闲置 10ms 后回收，任务队列为延时阻塞队列。
    应用场景：执行定时或周期性的任务。

*  使用实例

// 1. 创建 定时线程池对象 & 设置线程池线程数量固定为5
    ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
    // 2. 创建好Runnable类线程对象 & 需执行的任务
    Runnable task =new Runnable(){
        
      public void run() {
        
         System.out.println("执行任务啦");
      }
    };
    // 3. 向线程池提交任务
    scheduledThreadPool.schedule(task, 1, TimeUnit.SECONDS); // 延迟1s后执行任务
    scheduledThreadPool.scheduleAtFixedRate(task,10,1000,TimeUnit.MILLISECONDS);// 延迟10ms后、每隔1000ms执行任务

#### 2.5.3、可缓存线程池（CachedThreadPool） ####

*  源码

public static ExecutorService newCachedThreadPool() {
        
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
        
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>(),
                                      threadFactory);
    }
    
    
    特点：无核心线程，非核心线程数量无限，执行完闲置 60s 后回收，任务队列为不存储元素的阻塞队列。
    应用场景：执行大量、耗时少的任务。

*  使用实例

// 1. 创建可缓存线程池对象
    ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
    // 2. 创建好Runnable类线程对象 & 需执行的任务
    Runnable task =new Runnable(){
        
      public void run() {
        
         System.out.println("执行任务啦");
      }
    };
    // 3. 向线程池提交任务
    cachedThreadPool.execute(task);

#### 2.5.4、单线程化线程池（SingleThreadExecutor） ####

*  源码

public static ExecutorService newSingleThreadExecutor() {
        
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }
    
    
    特点：只有 1 个核心线程，无非核心线程，执行完立即回收，任务队列为链表结构的有界队列。
    应用场景：不适合并发但可能引起 IO 阻塞性及影响 UI 线程响应的操作，如数据库操作、文件操作等。

*  实例

// 1. 创建单线程化线程池
    ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
    // 2. 创建好Runnable类线程对象 & 需执行的任务
    Runnable task =new Runnable(){
        
      public void run() {
        
         System.out.println("执行任务啦");
      }
    };
    // 3. 向线程池提交任务
    singleThreadExecutor.execute(task);

#### 2.5.5、对比 ####

![20220810.png][]

### 2.6、总结 ###

> 1、Executors 的 4 个功能线程池虽然方便，但现在已经不建议使用了，而是建议直接通过使用 ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

> 2、FixedThreadPool 和 SingleThreadExecutor：主要问题是堆积的请求处理队列均采用 LinkedBlockingQueue，可能会耗费非常大的内存，甚至 OOM。

> 3、CachedThreadPool 和 ScheduledThreadPool：主要问题是线程数最大数是 Integer.MAX\_VALUE，可能会创建数量非常多的线程，甚至 OOM。

[adbdda1a828740fcbd7e93ffc737456f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/3cc2ddd8e73e4ee0b30eee48ab62998b.png
[f724be57e94c475b843e9d2a0d8f5bc4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/afcf58676d5f4be6a07ee83d1a946105.png
[2091e681eabc4a4ea4fc721be77a61a9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/52e161aec6724925894917494e7bc32e.png
[20220811.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/1aba25a3f2d34c8090944b89ed515e65.png
[459d0f554d5d4b5a34157f07e60cd27d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/93e75c4c37c34c66b6e0c7d94af739cc.png
[20220810.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/4ff7dae5c43e40498aad9e186aa82899.png