美文网首页大数据程序员技术干货
ThreadPoolExecutor源码解析(一)

ThreadPoolExecutor源码解析(一)

作者: 大数据_zzzzMing | 来源:发表于2018-06-01 14:44 被阅读62次

    1.ThreadPoolExcuter原理说明

      首先我们要知道为什么要使用ThreadPoolExcuter,具体可以看看文档中的说明:

      线程池可以解决两个不同问题:由于减少了每个任务的调用开销,在执行大量的异步任务时,它通常能够提供更好的性能,并且还可以提供绑定和管理资源(包括执行集合任务时使用的线程)的方法。每个 ThreadPoolExecutor还维护着一些基本的统计数据,如完成的任务数。

      线程池做的其实可以看得很简单,其实就是把你提交的任务(task)进行调度管理运行,但这个调度的过程以及其中的状态控制是比较复杂的。

    2.初始化参数介绍

    可以直接看最完整的ThreadPoolExcuter的初始化函数:

    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long

    keepAliveTime,TimeUnit unit,BlockingQueue workQueue,

    ThreadFactory threadFactory,

    RejectedExecutionHandler handler) {

    ...

    }

    逐个介绍如下:

    corePoolSize:核心线程数,在ThreadPoolExcutor中有一个与它相关的配置:allowCoreThreadTimeOut(默认为false),当allowCoreThreadTimeOut为false时,核心线程会一直存活,哪怕是一直空闲着。而当allowCoreThreadTimeOut为true时核心线程空闲时间超过keepAliveTime时会被回收。

    maximumPoolSize:最大线程数,线程池能容纳的最大线程数,当线程池中的线程达到最大时,此时添加任务将会采用拒绝策略,默认的拒绝策略是抛出一个运行时错误(RejectedExecutionException)。值得一提的是,当初始化时用的工作队列为LinkedBlockingDeque时,这个值将无效。

    keepAliveTime:存活时间,当非核心空闲超过这个时间将被回收,同时空闲核心线程是否回收受allowCoreThreadTimeOut影响。

    unit:keepAliveTime的单位。

    workQueue:任务队列,常用有三种队列,即SynchronousQueue,LinkedBlockingDeque(无界队列),ArrayBlockingQueue(有界队列)。

    threadFactory:线程工厂,ThreadFactory是一个接口,用来创建worker。通过线程工厂可以对线程的一些属性进行定制。默认直接新建线程。

    RejectedExecutionHandler:也是一个接口,只有一个方法,当线程池中的资源已经全部使用,添加新线程被拒绝时,会调用RejectedExecutionHandler的rejectedExecution法。

    默认是抛出一个运行时异常。

      这么多参数看起来好像很复杂,所以Java贴心得为我们准备了便捷的API,即可以直接用Executors创建各种线程池。分别是:

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

    //通过设置corePoolSize为0,而maximumPoolSize为Integer.Max_VALUE(Int型数据最大值)实现。

    ExecutorService cache =Executors.newCachedThreadPool();

    //创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。

    //通过将corePoolSize和maximumPoolSize的值设置为一样的值来实现。        ExecutorService fixed

    = Executors.newFixedThreadPool(num);

    //创建一个定长线程池,支持定时及周期性任务执行。

    //通过将队列参数workQueue设置为DelayWorkQueue来实现。        ExecutorService schedule

    = Executors.newScheduledThreadPool(5);

    //创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

    //通过将corePoolSize和maximumPoolSize都设置为1来实现。        ExecutorService single

    = Executors.newSingleThreadExecutor();

       这几个API会根据具体的情况而使用预设定好默认的初始化参数去创建一个ThreadPoolExecutor。

      这里需要做一个额外说明,在ThreadPoolExcuter中,worker和task是有区别的,task是用户提交的任务,而worker则是用来执行task的线程。在初始化参数中,corePoolSize和maximumPoolSize都是针对worker的,而workQueue是用来存放task的。

    3.worker介绍

      前面有介绍了一下worker和task的区别,其中task是用户提交的线程任务,而worker则是ThreadPoolExecutor自己内部实现的一个类了。

      具体源码如下:

    /** * Woker主要维护着运行task的worker的中断控制信息,以及其他小记录。这个类拓展AbstractQueuedSynchronizer

        * 而来简化获取和释放每一个任务执行中的锁。这可以防止中断那些打算唤醒正在等待其他线程任务的任务,而不是

        * 中断正在运行的任务。我们实现一个简单的不可重入锁而不是ReentrantLo,因为我们不想当其调用setCorePoolSize

        * 这样的方法的时候能获得锁。

        *///worker主要是对进行中的任务进行中断控制,顺带着对其他行为进行记录privatefinalclass Worker

                extends AbstractQueuedSynchronizer

                implements Runnable

        {

            /**        * This class will never be serialized, but we provide a

            * serialVersionUID to suppress a javac warning.

            */privatestaticfinallongserialVersionUID = 6138294804551838833L;

            /** Thread this worker is running in.  Null if factory fails. *///正在跑的线程,如果是null标识factory失败final Thread thread;

            /** Initial task to run.  Possibly null. *///初始化一个任务以运行        Runnable firstTask;

            /** Per-thread task counter *///每个线程计数volatilelong completedTasks;

            /**        * Creates with given first task and thread from ThreadFactory.

            * @param firstTask the first task (null if none)

            * 用给定的first task和从threadFactory创建

            */        Worker(Runnable firstTask) {

                setState(-1);// inhibit interrupts until runWorkerthis.firstTask = firstTask;

                this.thread = getThreadFactory().newThread(this);

            }

            /** Delegates main run loop to outer runWorker  *///主要调用了runWorkerpublicvoid run() {

                runWorker(this);

            }

            // Lock methods

            //// The value 0 represents the unlocked state.

            // The value 1 represents the locked state.

            //锁方法

            //protectedboolean isHeldExclusively() {

                returngetState() != 0;

            }

            //尝试获取锁protectedbooleantryAcquire(int unused) {

                if(compareAndSetState(0, 1)) {

                    setExclusiveOwnerThread(Thread.currentThread());

                    returntrue;

                }

                returnfalse;

            }

            //尝试释放锁protectedbooleantryRelease(int unused) {

                setExclusiveOwnerThread(null);

                setState(0);

                returntrue;

            }

            publicvoidlock()        { acquire(1); }

            publicbooleantryLock()  {returntryAcquire(1); }

            publicvoidunlock()      { release(1); }

            publicbooleanisLocked() {return isHeldExclusively(); }

            void interruptIfStarted() {

                Thread t;

                if(getState() >= 0 && (t = thread) !=null&& !t.isInterrupted()) {

                    try {

                        t.interrupt();

                    } catch (SecurityException ignore) {

                    }

                }

            }

        }

       Worker其实可以看作高级一点的线程。其中继承AbstractQueuedSynchronizer主要是为了实现锁控制。ThreadPoolExecutor会持有并管理Worker,在Worker中firstTask其实就是存放task的,而thread则是存放当前Worker本身的线程。

    其中比较重要的就是run方法了,但这个方法其实又是去调用ThreadPoolExecutor里面的runWorker()方法,具体可以看下一节的介绍。

    4.ctl介绍以及运行状态说明

    首先需要介绍线程池有五种运行状态:

    RUNNING(状态值-1): 接收新任务并处理队列中的任务

    SHUTDOWN(状态值0): 不接收新任务但会处理队列中的任务。

    STOP(状态值1): 不接收新任务,不处理队列中的任务,并中断正在处理的任务

    TIDYING(状态值2): 所有任务已终止,workerCount为0,处于TIDYING状态的线程将调用钩子方法terminated()。

    TERMINATED(状态值3): terminated()方法完成。

    然后我们可以看看ThreadPoolExcuter中的ctl这个变量。

    ctl是ThreadPoolExcuter中比较有意思的一个实现,它是一个AtomicInteger,这里不对AtomicInteger多做讨论,只要知道可以把它看成有原子性的Integer就够了,其实它具有原子性的原理是使用了CAS的技术,这是一种乐观锁的具体实现。

    ThreadPoolExcuter是将两个内部值打包成一个值,即将workerCount和runState(运行状态)这两个值打包在一个ctl中,因为runState有5个值,需要3位,所以有3位表示

    runState,而其他29位表示为workerCount。

    而运行时要获取其他数据时,只需要对ctl进行拆包即可。具体这部分代码如下:

    privatefinalAtomicInteger ctl =newAtomicInteger(ctlOf(RUNNING, 0));

        privatestaticfinalintCOUNT_BITS = Integer.SIZE - 3;

        privatestaticfinalintCAPACITY  = (1 << COUNT_BITS) - 1;

        // runState is stored in the high-order bitsprivatestaticfinalintRUNNING    = -1 << COUNT_BITS;

        privatestaticfinalintSHUTDOWN  =  0 << COUNT_BITS;

        privatestaticfinalintSTOP      =  1 << COUNT_BITS;

        privatestaticfinalintTIDYING    =  2 << COUNT_BITS;

        privatestaticfinalintTERMINATED =  3 << COUNT_BITS;

        //Packing and unpacking ctl

    //拆包ctl,分别获取runState和WorkerCount

    privatestaticintrunStateOf(intc)    {returnc & ~CAPACITY; }

        privatestaticintworkerCountOf(intc)  {returnc &CAPACITY; }

    //打包操作

    privatestaticintctlOf(intrs,intwc) {returnrs | wc; }

    5.拒绝策略

    当执行器(Executor)处于终止状态,或者执行器在max threads和工作队列都是有界并且处于饱和的时候,新提交的任务会被拒绝。在任一情况下,执行的任务将调用RejectedExecutionHandler的方法rejectedExecution(Runnable, ThreadPoolExecutor)。有以下四种拒绝策略:

    1.默认的是ThreadPoolExecutor.AbortPolicy,在这种策略下,处理器会在拒绝后抛出一个运行异常RejectedExecutionException。

    2.在ThreadPoolExecutor.CallerRunsPolicy的策略下,线程会调用它直接的execute来运行这个任务。这种方式提供简单的反馈控制机制来减缓新任务提交的速度。

    3.在ThreadPoolExecutor.DiscardPolicy策略下,无法执行的任务将被简单得删除掉。

    4.在ThreadPoolExecutor.DiscardOldestPolicy策略下,如果executor没有处于终止状态,在工作队列头的任务将被删除,然后会重新执行(可能会再次失败,这会导致重复这个过程)。

    总结:本篇初步介绍了ThreadPoolExcuter的基本原理,解决了什么问题。而后说明了ThreadPoolExcuter中的初始化参数,对其中的各个参数做初步介绍。再之后Worker以及ctl变量的作用,并初步介绍了任务提交失败后的拒绝策略。

    相关文章

      网友评论

      本文标题:ThreadPoolExecutor源码解析(一)

      本文链接:https://www.haomeiwen.com/subject/fcqksftx.html