详述Java线程池实现原理

Posted 李子捌

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了详述Java线程池实现原理相关的知识,希望对你有一定的参考价值。

一、写在前面

1.1 线程池是什么

线程池(Thread Pool) 是一种池化思想管理线程的工具,经常出现在多线程服务器中,如mysql。

线程过多会带来额外的开销,其中包括创建销毁线程的开销,操作系统调度线程的开销等等,同时也降低了计算机的整体性能。线程池维护多个线程,等待监督管理者分配可并发执行的任务。这种做法,一方面避免了处理任务是创建销毁线程开销代价,另一方面避免了线程数量膨胀导致的过分调度问题,保证了对操作系统内核的充分利用。

本文描述的线程池是JDK提供的ThreadPoolExecutor类

使用线程池带来的好处

  • 降低资源消耗:通过赤化技术重复利用已创建的线程,降低想成创建和 销毁造成的消耗
  • 提高响应速度:任务到达时,无需等待线程创建即可立即执行
  • 提高线程的可管理性:线程是稀缺资源,如果无限制创建,不仅会消耗系统资源,还会因为线程的不合理分配导致资源调度失衡,降低系统的稳定性。使用线程池可以进行统一的分配、调优和监控
  • 提供更多更强大的功能:线程池具备可拓展性,允许开发人员向其中增加风多的功能。比如延时定时线程池ScheduledThreadPoolExecutor,就允许任务延期执行或定期执行

1.2 线程池解决的问题是什么

线程池解决的问题就是资源管理的问题。在并发环境下,系统不能够确定在任意时刻有多少任务需要执行,有多少资源需要投入。

在这种不确定性下将会带来以下若干的问题

  • 频繁申请/销毁资源和调度资源,将带来额外的开销,可能是非常巨大的
  • 对资源无限申请缺少抑制手段,容易引发系统资源耗尽问题的风险
  • 系统无法合理管理内部的资源分布,会减低系统的稳定性

为了解决资源分配的问题,线程池采用“池化”(Pooling)思想。池化,顾名思义,是为了做大化收益并最小化风险,而将资源统一在一起管理的一种思想。

在计算机领域池化技术表现为:统一管理IT资源,包括服务器资源、存储、网络资源等。通过共享资源,使用户在第投入中获益。

除去线程池其他比较典型的几种使用策略包括

  • 内存池(Memory Pooling):预先申请内存,提升申请内存的速度,减少内存碎片
  • 连接池(Connection Pooling):预先申请数据库连接,提升申请连接的速度,降低系统开销
  • 实例池(Object Pooling):循环使用对象,减资源在初始化和释放时昂贵的损耗

二、线程池和核心设计与实现

2.1 总体设计

Java中线程池核心实现类是ThreadPoolExecutor,本章基于JDK1.8的源码来分析Java线程池的核心设计与实现。首先看一下ThreadPoolExecutor的UML图,了解ThreadPoolExecutor的继承关系

ThreadPoolExecutor实现的顶层接口是Executor,顶层接口Executor提供了一种思想:将任务提交和任务执行进行解耦。用户无需关注如何创建线程,如何调度线程来执行任务,用户只需提供Runnable对象,将任务的运行逻辑提交到执行器Executor中,由Executor框架完成线程的调配和任务的执行部分。

ExecutorService

  • 扩充执行任务的能力,补充可以为一个或者一批异步任务生成Future的方法
  • 提供了管理线程池的方法,比如停止线程池的运行

AbstractExecutorService

  • 串联任务流程,保证下层的实现只需要关注一个执行任务的方法

ThreadPoolExecutor

  • 维护自身的生命周期
  • 管理线程和任务,使两者良好的结合从而执行并行任务

ThreadPoolExecutor是如何运行,如何同时维护线程和执行任务的呢?其运行机制如下图所示

ThreadPoolExecutor运行流程

线程池在内部实际上构造了一个生产者消费者模型,将线程和任务两者解耦,并不直接关联,从而良好的管理缓冲任务,复用线程。线程池的运行主要分成两部分::任务管理、线程管理。任务管理充当生产者角色,当任务提交后,线程池会判断该任务后续流转

  • 任务申请线程执行该任务
  • 缓冲到队列中等待线程执行
  • 拒绝该任务

线程管理部分是消费者,它们被统一维护在线程池内,根据任务请求进行线程的分配,当线程执行完任务后会继续获取新的任务执行,最终获取不到任务的时候,线程会被回收。

接下来按照如下三个方面讲解线程池的运行机制:

  • 线程池如何维护自身状态
  • 线程池如何管理任务
  • 线程池如何管理线程

2.2 生命周期管理

线程池运行的状态,并不是用户显式设置的,而是伴随着线程池的运行,由内部来维 护。线程池内部使用一个变量维护两个值:运行状态 (runState) 和线程数量 (workerCount)。在具体实现中,线程池将运行状态 (runState)、线程数量 (workerCount)

两个关键参数的维护放在了一起,如下代码所示:

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

ctl 这个 AtomicInteger 类型,是对线程池的运行状态和线程池中有效线程的数量 进行控制的一个字段,它同时包含两部分的信息:线程池的运行状态 (runState) 和 线程池内有效线程的数量 (workerCount),高 3 位保存 runState,低 29 位保存 workerCount,两个变量之间互不干扰。用一个变量去存储两个值,可避免在做相关 决策时,出现不一致的情况,不必为了维护两者的一致,而占用锁资源。通过阅读线 程池源代码也可以发现,经常出现要同时判断线程池运行状态和线程数量的情况。线程池也提供了若干方法去供用户获得线程池当前的运行状态、线程个数。这里都使用的是位运算的方式,相比于基本运算,速度也会快很多

关于内部封装的获取生命周期状态、获取线程池线程数量的计算方法如以下代码 所示:

// Packing and unpacking ctl
// 计算当前运行状态
private static int runStateOf(int c)      return c & ~CAPACITY;  
// 计算当前线程数据
private static int workerCountOf(int c)   return c & CAPACITY; 
// 通过状态和线程数生成ctl
private static int ctlOf(int rs, int wc)  return rs | wc; 

ThreadPoolExecutor 的运行状态有 5 种,分别为:

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;
运行状态状态描述
RUNNING能接受新提交的任务,并且也能处理阻塞队列中的任务
SHUTDOWN状态关闭,不在接受新提交的任务,但是能继续处理阻塞队列已保存的让任务
STOP不接受新任务,也不处理队列中的任务,会中断正在处理任务的线程
TIDYING所有让任务都已终止,workerCount(有效处理让任务线程)状态为0
TERMINATED在terminated()方法执行结束后进入该状态

其生命周期转换如下图所示

线程池生命周期

2.3 任务调度机制

2.3.1 任务调度

任务调度是线程池的主要入口,当用户提交了一个任务,接下来这个任务将如何执行 都是由这个阶段决定的。了解这部分就相当于了解了线程池的核心运行机制。 首先,所有任务的调度都是由 execute 方法完成的,这部分完成的工作是:检查现在线程池的运行状态、运行线程数、运行策略,决定接下来执行的流程,是直接申请线程执行,或是缓冲到队列中执行,亦或是直接拒绝该任务。其执行过程如下:

  • 首先检测线程池运行状态,如果不是 RUNNING,则直接拒绝,线程池要保证在 RUNNING 的状态下执行任务
  • 如果 workerCount < corePoolSize,则创建并启动一个线程来执行新提交的任务
  • 如果 workerCount >= corePoolSize,且线程池内的阻塞队列未满,则将任务添加到该阻塞队列中。
  • 如 果 workerCount >= corePoolSize && workerCount < maximumPoolSize,且线程池内的阻塞队列已满,则创建并启动一个线程来执行新提交的任务。
  • 如果 workerCount >= maximumPoolSize,并且线程池内的阻塞队列已满 , 则根据拒绝策略来处理该任务 , 默认的处理方式是直接抛异常。

其执行流程如下

任务调度流程图

2.3.2 任务缓冲

任务缓冲模块是线程池能够管理任务的核心部分。线程池的本质是对任务和线程的管 理,而做到这一点最关键的思想就是将任务和线程两者解耦,不让两者直接关联,才 可以做后续的分配工作。线程池中是以生产者消费者模式,通过一个阻塞队列来实现 的。阻塞队列缓存任务,工作线程从阻塞队列中获取任务。

阻塞队列 (BlockingQueue) 是一个支持两个附加操作的队列。这两个附加的操作是: 在队列为空时,获取元素的线程会等待队列变为非空。当队列满时,存储元素的线程 会等待队列可用。阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元 素的线程,消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素。

阻塞队列

使用不同的队列可以实现不一样的任务存取策略。在这里,我们可以再介绍下阻塞队列的成员:

名称描述
ArrayBlockingQueue一个用数组实现的有界阻塞队列,此队列按照先进先出(FIFO)的原则对元素进行排序。支持公平锁和非公平锁
LinkedBlockingDeque一个由链表结构组成的有界队列,此队列按照先进先出(FIFO)的原则对元素进行排序。此队列的默认长度为Integer.MAX_VALUE,所以默认创建此队列有容量危险
PriorityBlockingQueue一个支持线程优先级排序的无界队列,默认自然进行排序,也可以自定义实现compareTo()方法指定排序故障,不能保证同优先级元素的顺序。
DelayQueue一个实现PriorityBlockingQueue实现延迟获取的无界队列,在创建元素时,可以指定多久才能从队列中获取当前元素。只有延迟期满后才能从队列中获取元素。
SynchronousQueue一个不存储元素的阻塞队列,每个put操作必须等待take操作,否则不能添加元素。支持公平锁和非公平锁。SynchronousQueue的一个使用场景是在线程池里。Executors.newCachedThreadPool()就使用了SynchronousQueue,这个线程池根据需要(新任务来)创建新的线程,如果有空闲的线程就使用空闲线程,线程空闲60秒会被回收。
return new ThreadPoolExecutor(
0,
Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue());
LinkedTransferQueue一个由链表结构组成的无界阻塞队列,相当于其他队列,LinkedTransferQueue多了transfer和tryTransfer方法
LinkedBlockingQueue一个由链表结构组成的双向阻塞队列,队列的头部和尾部都可以插入和删除元素,多线程并发时,可以将锁的竞争最多降到一半

2.3.3 任务申请

由上文的任务分配部分可知,任务的执行有两种可能:一种是任务直接由新创建的线 程执行。另一种是线程从任务队列中获取任务然后执行,执行完任务的空闲线程会再 次去从队列中申请任务再去执行。第一种情况仅出现在线程初始创建的时候,第二种 是线程获取任务绝大多数的情况。

线程需要从任务缓存模块中不断地取任务执行,帮助线程从阻塞队列中获取任务,实现线程管理模块和任务管理模块之间的通信。这部分策略由 getTask 方法实现,其 执行流程如下图所示:

线程获取任务的流程

getTask 这部分进行了多次判断,为的是控制线程的数量,使其符合线程池的状 态。如果线程池现在不应该持有那么多线程,则会返回 null 值。工作线程 Worker 会不断接收新任务去执行,而当工作线程 Worker 接收不到任务的时候,就会开始 被回收。

源码分析

private Runnable getTask() 
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) 
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        // 判断线程池是否已停止运行
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) 
            decrementWorkerCount();
            return null;
        

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
		// 判断线程现阶段是否够多
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) 
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        
		// 限时任务获取和阻塞获取
        try 
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
            workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
         catch (InterruptedException retry) 
            timedOut = false;
        
    

2.3.4 任务拒绝

任务拒绝模块是线程池的保护部分,线程池有一个最大的容量,当线程池的任务缓存 队列已满,并且线程池中的线程数目达到 maximumPoolSize 时,就需要拒绝掉该任务,采取任务拒绝策略,保护线程池。

拒绝策略是一个接口,其设计如下:

public interface RejectedExecutionHandler 

    /**
     * Method that may be invoked by a @link ThreadPoolExecutor when
     * @link ThreadPoolExecutor#execute execute cannot accept a
     * task.  This may occur when no more threads or queue slots are
     * available because their bounds would be exceeded, or upon
     * shutdown of the Executor.
     *
     * <p>In the absence of other alternatives, the method may throw
     * an unchecked @link RejectedExecutionException, which will be
     * propagated to the caller of @code execute.
     *
     * @param r the runnable task requested to be executed
     * @param executor the executor attempting to execute this task
     * @throws RejectedExecutionException if there is no remedy
     */
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);

用户可以通过实现这个接口去定制拒绝策略,也可以选择 JDK 提供的四种已有拒绝策略,其特点如下

名称描述
ThreadPoolExecutor.AbortPolicy丢弃任务并抛出RejectedExecutionException异常。这是线程池默认的拒绝策略,在任务不能在提交的时候,抛出异常,及时反馈程序运行状态。如果是比较关键的业务,推荐使用该策略,这样子在系统不能承载更大并发的时候,能过及时的通过异常发现。
ThreadPoolExecutor.DiscardPolicy丢弃任务,但是不抛出异常。使用该策略,可能会使我们无法发现系统的异常状态。建议一些无关紧要的业务采用此策略。
ThreadPoolExecutor.DiscardOldestPolicy丢弃队列最前面的任务,然后重新提交比拒接的任务。是否要采用此种策略,需要根据实际业务是否允许丢弃老任务来认真衡量
ThreadPoolExecutor.CallerRunsPolicy由调用线程(提交任务的线程)来处理任务。这种情况是需要让所有的任务都执行完毕,那么就适合大量计算的任务类型去执行,多线程仅仅是增加大吞吐量的手段,最终必须要让每个任务都执行
/**
     * A handler for rejected tasks that runs the rejected task
     * directly in the calling thread of the @code execute method,
     * unless the executor has been shut down, in which case the task
     * is discarded.
     */
public static class CallerRunsPolicy implements RejectedExecutionHandler 
    /**
         * Creates a @code CallerRunsPolicy.
         */
    public CallerRunsPolicy()  

    /**
         * Executes task r in the caller's thread, unless the executor
         * has been shut down, in which case the task is discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
        if (!e.isShutdown()) 
            r.run();
        
    


/**
     * A handler for rejected tasks that throws a
     * @code RejectedExecutionException.
     */
public static class AbortPolicy implements RejectedExecutionHandler 
    /**
         * Creates an @code AbortPolicy.
         */
    public AbortPolicy()  

    /**
         * Always throws RejectedExecutionException.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         * @throws RejectedExecutionException always
         */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    


/**
     * A handler for rejected tasks that silently discards the
     * rejected task.
     */
public static class DiscardPolicy implements RejectedExecutionHandler 
    /**
         * Creates a @code DiscardPolicy.
         */
    public DiscardPolicy()  

    /**
         * Does nothing, which has the effect of discarding task r.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
    


/**
     * A handler for rejected tasks that discards the oldest unhandled
     * request and then retries @code execute, unless the executor
     * is shut down, in which case the task is discarded.
     */
public static class DiscardOldestPolicy implements RejectedExecutionHandler 
    /**
         * Creates a @code DiscardOldestPolicy for the given executor.
         */
    public DiscardOldestPolicy()  

    /**
         * Obtains and ignores the next task that the executor
         * would otherwise execute, if one is immediately available,
         * and then retries execution of task r, unless the executor
         * is shut down, in which case task r is instead discarded.
         *
         * @param r the runnable task requested to be executed
         * @param e the executor attempting to execute this task
         */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) 
        if (!e.isShutdown()) 
            e.getQueue().poll();
            e.execute(r);
        
    

2.4 Worker线程管理

2.4.1 Worker线程

线程池为了掌握线程的状态并维护线程的生命周期,设计了线程池内的工作线程Worker。

Java Worker源码部分

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    
        /** Thread this worker is running in.  Null if factory fails. */
        // worker持有的线程
        final Thread thread;
        /** Initial task to run.  Possibly null. */
        // 初始化的任务,可以为null
        Runnable firstTask;
    
        ...

Worker 这个工作线程,实现了 Runnable 接口,并持有一个线程 thread,一个初始化的任务 firstTask。thread 是在调用构造方法时通过 ThreadFactory 来创建的线程,可以用来执行任务;firstTask 用它来保存传入的第一个任务,这个任务可以有也可以为 null。如果这个值是非空的,那么线程就会在启动初期立即执行这个任务,也就对应核心线程创建时的情况;如果这个值是 null,那么就需要创建一个线程去执行任务列表(workQueue)中的任务,也就是非核心线程的创建

/**
     * The queue used for holding tasks and handing off to worker
     * threads.  We do not require that workQueue.poll() returning
     * null necessarily means that workQueue.isEmpty(), so rely
     * solely on isEmpty to see if the queue is empty (which we must
     * do for example when deciding whether to transition from
     * SHUTDOWN to TIDYING).  This accommodates special-purpose
     * queues such as DelayQueues for which poll() is allowed to
     * return null even if it may later return non-null when delays
     * expire.
     */
# workerQueue 源码定义
private final BlockingQueue<Runnable> workQueue;

worker执行任务

线程池需要管理线程的生命周期,需要在线程长时间不运行的时候进行回收。线程池 使用一张 Hash 表去持有线程的引用,这样可以通过添加引用、移除引用这样的操作 来控制线程的生命周期。这个时候重要的就是如何判断线程是否在运行。

/**
     * Set containing all worker threads in pool. Accessed only when
     * holding mainLock.
     */
private final HashSet<Worker> workers = new HashSet<Worker>();

Worker 是通过继承 AQS,使用 AQS 来实现独占锁这个功能。没有使用可重入锁ReentrantLock,而是使用 AQS,为的就是实现不可重入的特性去反应线程现在的执行状态。

private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
  • lock方法一旦获取了独占锁,表示当前线程正在执行任务中
  • 如果正在执行任务,则不应该中断线程
  • 如果该线程现在不是独占锁状态,也就是空闲状态,说明它没有正在处理任务,这时可以对该线程进行中断
  • 线程池在执行shutdown方法或tryTeriminate方法是或调用interruptIdleWorkers方法来中断空闲线程,interruptIdleWorkers方法会使用tryLock方法来判断线程池中的线程是否是空闲状态,如果是空闲状态则可以安全回收

shutdown方法源码

 public void shutdown() 
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try 
            checkShutdownAccess();
            advanceRunState(SHUTDOWN);
            // 执行interruptIdleWorkers方法
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
         finally 
            mainLock.unlock();
        
        tryTerminate();
    

tryTerminate方法源码

final void tryTerminate() 
        for (;;) 
            int c = ctl.get();
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                return;
            if (workerCountOf(c) != 0)  // Eligible to terminate
                // 执行interruptIdleWorkers
                interruptIdleWorkers(ONLY_ONE);
                return;
            

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try 
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) 
                    try 
                        terminated();
                     finally 
                        ctl.set(ctlOf(TERMINATED, 0));
                        termination.signalAll();
                    
                    return;
                
             finally 
                mainLock.unlock();
            
            // else retry on failed CAS
        
    

interruptIdleWorkers方法源码

 private void interruptIdleWorkers(boolean onlyOne) 
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try 
            for (Worker w : workers) 
                Thread t = w.thread;
                if (!t.isInterrupted() && w.tryLock()) 
                    try 
                        t.interrupt();
                     catch (SecurityException ignore) 
                     finally 
                        w.unlock();
                    
                
                if (onlyOne)
                    break;
            
         finally 
            mainLock.unlock();
        
    

在线程回收过程中就使用到了这种特性,回收过程如下图所示:

线程池回收过程

2.4.2 worker线程增加

增加线程是通过线程池中的 addWorker 方法,该方法的功能就是增加一个线程, 该方法不考虑线程池是在哪个阶段增加的该线程,这个分配线程的策略是在上个步 骤完成的,该步骤仅仅完成增加线程,并使它运行,最后返回是否成功这个结果。 addWorker 方法有两个参数:firstTask、core。firstTask 参数用于指定新增的线程执行的第一个任务,该参数可以为空;core 参数为 true 表示在新增线程时会判断当前活动线程数是否少于 corePoolSize,false 表示新增线程前需要判断当前活动线程数是否少于 maximumPoolSize,其执行流程如下图所示:

申请线程执行流程图

源码分析

 private boolean addWorker(Runnable firstTask, boolean core) 
        retry:
        for (;;) 
            int c = ctl.get();
            int rs = runStateOf(c);

            // 判断线程是否已经停止
            // 判断线程是否正在停止 如果是则判断线程是否用于执行剩余任务firstTask
            // workQueue是否为空
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) 
                // 获取线程数量
                int wc = workerCountOf(c);
                // 判断线程是否超过容量 
                // 判断线程是否超过对应核心数  上面讲了core 传true/false区别
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
              	
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs以上是关于详述Java线程池实现原理的主要内容,如果未能解决你的问题,请参考以下文章

java 线程池机制的原理是啥?

Java线程池的实现原理,你清楚么?

聊聊面试中的 Java 线程池

阶段1 语言基础+高级_1-3-Java语言高级_05-异常与多线程_第5节 线程池_1_线程池的概念和原理

深入源码,深度解析Java 线程池的实现原理

深入源码分析Java线程池的实现原理