JDK核心源码之FixedThreadPool源码注释

    今天，咱们来分析一下并发里面关于线程池的原理，咱们经过源码深刻剖析，这样了解到其中的细节，对于咱们把握好并发的线程池使用很是有帮助。

    你们都知道线程池的类型有好几个，可是咱们只要搞清楚其中的一个，其他的其实很好分析。这儿就以FixedThreadPool举例。从源码的角度去理解这个线程池。

首先先建立一个线程池//建立线程池ExecutorService executorService = Executors.newFixedThreadPool(4);        executorService.execute(new Runnable() {            public void run() {                System.out.println("test");            }        });跟踪newFixedThreadPool方法，代码以下：    public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());​}

 

很显然咱们看到这个是重要的方法是类ThreadPoolExecutor的，点进去：

 

//corePoolSize:核心线程数//maximumPoolSize:线程池里最多能够建立多少线程//keepAliveTime:线程能够存活多久//TimeUnit: 时间的单位，配合keepAliveTime一块儿使用//workQueue: 若是线程的个数超过核心线程数的数量了，就会把多出来的线程放入到这个队列，若是这个队列也存满了，而后就会在建立出线程，可是总的线程不能多于maximumPoolSize的值。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,                           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.acc = System.getSecurityManager() == null ? null :          AccessController.getContext();          this.corePoolSize = corePoolSize;          this.maximumPoolSize = maximumPoolSize;          this.workQueue = workQueue;          this.keepAliveTime = unit.toNanos(keepAliveTime);          this.threadFactory = threadFactory;          this.handler = handler;​}

 

都分析到构造函数的位置了，那么按照咱们通常分析源码的习惯，确定须要看一下，这个类里面主要的成员变量和核心方法。以下变量须要咱们注意：

//这个变量很是的重要，//高3位表明这线程的状态//低29位表明着线程的数量private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//29private static final int COUNT_BITS = Integer.SIZE - 3;// 1 << 29 - 1  = 00011111 11111111 11111111 11111111//1^29 - 1;private static final int CAPACITY   = (1 << COUNT_BITS) - 1;// 线程池的状态// 11100000 00000000 00000000 00000000private static final int RUNNING    = -1 << COUNT_BITS;//获取线程的状态private static int runStateOf(int c)     {return c & ~CAPACITY;}//获取线程的个数private static int workerCountOf(int c)  {return c & CAPACITY;}private static int ctlOf(int rs, int wc) {return rs | wc;   }​​ 

如上，咱们已经分析完了线程池的初始化工做。接下来咱们分析一下execute()方法。

ExecutorService executorService = Executors.newFixedThreadPool(10);//接下来咱们就要分析这个方法//这个是抽象方法，咱们看其实现类//ThreadPoolException里面的excute方法//这个方法很是重要executorService.execute(new Runnable() {public void run() {System.out.println("test");}});

 

 

  进到ThreadPoolException之后，咱们在里面找到execute方法：

  

//这个方法很是很是关键//并且下面的注释也是很是很是的重要。public void execute(Runnable command) {        if (command == null)            throw new NullPointerException();        /*         * Proceed in 3 steps:         1. If fewer than corePoolSize threads are running, try to         start a new thread with the given command as its first         * task.  The call to addWorker atomically checks runState and         * workerCount, and so prevents false alarms that would add         * threads when it shouldn't, by returning false.           1. 若是正在运行的线程数小于核心线程数，那么就尝试启动一个新的线程。              经过调用addWorker方法建立新的线程，建立线程的时候须要检测一下线程的状态              和线程总数         * 2. If a task can be successfully queued, then we still need         * to double-check whether we should have added a thread         * (because existing ones died since last checking) or that         * the pool shut down since entry into this method. So we         * recheck state and if necessary roll back the enqueuing if         * stopped, or start a new thread if there are none.           2.若是发现正在运行的线程的个数已经大于corePoolSize,那么就把当前的线程加入到           队列中。         * 3. If we cannot queue task, then we try to add a new         * thread.  If it fails, we know we are shut down or saturated         * and so reject the task.           3. 若是队列里面也放不下了，那么就会启动新的线程，可是总的线程的个数不能大于            maximumPoolSize         */        int c = ctl.get();        //第一步：若是当前线程的个数 小于 核心线程数        if (workerCountOf(c) < corePoolSize) {            //经过addWorker的方式启动线程            //true表明的是建立的是核心线程            if (addWorker(command, true))                return;            c = ctl.get();        }        //第二步：若是代码走到这儿，说明当前线程的个数已经大于corePoolSize了        //就把线程加入到队列里面去        //注意其实offer这个方法是不阻塞的        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);        }        //第三步：代码走到这儿说明队列存满了        //要建立新的线程，可是总的线程不能大于maximumPoolSize        //false表明的是建立非核心线程。        else if (!addWorker(command, false))            reject(command);}​​

     上面的核心代码画成图以下：

 

   咱们看了execute()方法的三个步骤之后，咱们接下来一个步骤一个步骤的去分析，你们注意到第一步和第三步 建立线程的时候都调用的是addWorker的方法，不一样的是传进去的参数一个是true，一个是false，true表明建立的是核心线程，false表明建立的是非核心线程。

   首先分析第一步的建立核心线程的addWorker方法，点进去会观看到以下代码：

   

 

private boolean addWorker(Runnable firstTask, boolean core) {        retry:        for (;;) {            int c = ctl.get();            //获取到线程池状态            int rs = runStateOf(c);            // Check if queue empty only if necessary.            //这儿能够不用多管就是检查一下            if (rs >= SHUTDOWN &&                ! (rs == SHUTDOWN &&                   firstTask == null &&                   ! workQueue.isEmpty()))                return false;            for (;;) {                //获取当前的线程数                int wc = workerCountOf(c);                //若是当前线程数大于总的线程数或者                //当前的线程大于corePoolSize 或者大于maximumPoolSize               //都返回false                //当前core是ture 因此该判断是： wc >= corePoolSize                //因此是若是当前线程数大于设置的核心线程数九返回false                if (wc >= CAPACITY ||                    wc >= (core ? corePoolSize : maximumPoolSize))                    return false;                //增长当前的线程数                //增长成功就跳出当前循环                //使用CAS的方式增长了线程数，保证了线程安全                if (compareAndIncrementWorkerCount(c))                    break retry;                c = ctl.get();  // Re-read ctl                if (runStateOf(c) != rs)                    continue retry;                // else CAS failed due to workerCount change; retry inner loop            }        }        boolean workerStarted = false;        boolean workerAdded = false;        Worker w = null;        try {            //Worker自己是一个线程            //里面还封装了一个线程            w = new Worker(firstTask);            final Thread t = w.thread;            if (t != null) {                //这儿加了锁                //这儿为何加锁，其实咱们想一下就明白了                // 咱们建立了一个线程对象（worker）,而后要把这个worker                // 放入到workers里面，那么这个过程咱们要保证线程安全，因此                //这个地方须要加锁                final ReentrantLock mainLock = this.mainLock;                mainLock.lock();                try {                    // Recheck while holding lock.                    // Back out on ThreadFactory failure or if                    // shut down before lock acquired.                    //获取线程池的状态                    int rs = runStateOf(ctl.get());                    //若是状态正常                    if (rs < SHUTDOWN ||                        (rs == SHUTDOWN && firstTask == null)) {                        if (t.isAlive()) // precheck that t is startable                            throw new IllegalThreadStateException();                        //把当前封装的线程加入到workers里面                        workers.add(w);                        int s = workers.size();                        if (s > largestPoolSize)                            largestPoolSize = s;                        workerAdded = true;                    }                } finally {                    //释放锁                    mainLock.unlock();                }                //若是正常执行的话，代码会走到这儿                if (workerAdded) {                    //调用t的start方法                    //这个方法较为重要，接下来咱们就是分析这个方法                   //这儿值得咱们注意的是这个t是Worker里面的t                   //因此咱们要想知道这个t.start()最终调的什么方法                   //咱们须要去看一下worker的内部                    t.start();                   workerStarted = true;                }            }        } finally {            if (! workerStarted)                addWorkerFailed(w);        }        return workerStarted;    }​

  接下来咱们分析一下t.start方法

  可是咱们从以前的代码得知，咱们是经过这个以下代码获取到的t:

  

//fristTask就是咱们传进来的代码w = new Worker(firstTask);final Thread t = w.thread;

   想知道t是啥，那么咱们进入看一下Worker的构造函数：

   

Worker(Runnable firstTask) {   setState(-1); // inhibit interrupts until runWorker    this.firstTask = firstTask;    //这个实际上是使用默认的工程建立的一个线程   //this就是Worker本身，Worker自己是实现了Runnable接口，因此他本身   //也是一个线程。因此到这个时候咱们就应该明白了。   //前面的t.start方法，其实调用的就是worker的run()方法    this.thread = getThreadFactory().newThread(this);}

 原来t是一个worker默认的线程工厂建立的线程，这样咱们就得知t.start其实调用的是Worker的run方法，接下来咱们分析一下，Worker的run方法。

  

/** Delegates main run loop to outer runWorker  */public void run() {    runWorker(this);}

 调用runWorker的方法，跟踪进去：

  

final void runWorker(Worker w) {    Thread wt = Thread.currentThread();    //这个是咱们传进来的代码    Runnable task = w.firstTask;    w.firstTask = null;    w.unlock(); // allow interrupts    boolean completedAbruptly = true;    try {        //一开始咱们建立线程的时候，这个时候task不等于null       //可是若是核心线程里面已经把task运行完了，那么这个时候task就为null       //若是task为null了，那么就会走getTask这个方法      //这个方法就会从无界队列里面获取task来执行        while (task != null || (task = getTask()) != null) {            w.lock();            // If pool is stopping, ensure thread is interrupted;            // if not, ensure thread is not interrupted.  This            // requires a recheck in second case to deal with            // shutdownNow race while clearing interrupt            if ((runStateAtLeast(ctl.get(), STOP) ||                (Thread.interrupted() &&                runStateAtLeast(ctl.get(), STOP))) &&                !wt.isInterrupted())                wt.interrupt();            try {                beforeExecute(wt, task);                Throwable thrown = null;                try {                    //让代码运行                    task.run();                } catch (RuntimeException x) {                    thrown = x; throw x;                } catch (Error x) {                    thrown = x; throw x;                } catch (Throwable x) {                    thrown = x; throw new Error(x);                } finally {                    afterExecute(task, thrown);                }            } finally {                task = null;                //统计信息                w.completedTasks++;                w.unlock();            }        }        completedAbruptly = false;    } finally {         //异常处理        processWorkerExit(w, completedAbruptly);    }}

到此为止，咱们分析到了前面说的三个不步骤的第一步，使用核心线程启动任务（咱们后面才分析启动任务之后，代码是如何运行的？）。接下来咱们分析一下第二步，如何把在核心线程满了之后的新任务压入队列。咱们回过头来看这个方法：

ThreadPoolExcutor的execute方法，贴出第二步的关键代码：

//若是线程池的状态是正在运行，那么就把咱们写的线程压入到 //orkerQueue里面，由于咱们分析的是FixedThreadPool线程池 //因此这个workerQueue就是LinkedBlockingQueue,并且这个队列默认的大小给了一个Integer的最大值。 //接下来咱们分析一下这个offer方法，不过这儿值得咱们注意的是 //虽然咱们知道LinkedBlockingQueue是一个阻塞队列，可是其offer方法是不阻塞的。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);}

 咱们能够进去看一下offer方法：

public boolean offer(E e) {    if (e == null) throw new NullPointerException();    final AtomicInteger count = this.count;    //重要，这个的意思就是，若是队列满了就返回false    //可是咱们的这个队列很大，咱们能够暂时认为通常状况下是不会满的。    if (count.get() == capacity)        return false;    int c = -1;    Node<E> node = new Node<E>(e);    final ReentrantLock putLock = this.putLock;    putLock.lock();    try {         //若是队列里面的线程数小于容量        if (count.get() < capacity) {            //压入队列            enqueue(node);            //当前存入的任务个数增长1            c = count.getAndIncrement();            //若是当前容量还能存数据            if (c + 1 < capacity)            //唤醒线程，其实咱们以前说过offer是没有阻塞功能的           //这儿的singnal主要是是配合take去使用，take方法有阻塞功能。                notFull.signal();        }    } finally {        putLock.unlock();    }    if (c == 0)        signalNotEmpty();    return c >= 0;}

    那接下来咱们分析ThreadPoolExcutor的execute方法里面的第三步，贴出关键源码以下：

//这方法咱们知道，由于前面的两个条件都不知足，也就是，核心线程也占用满了//队列也存满了，因此代码就会走到这儿。//这儿进去走的是addWorker的方法，其实这个方法咱们前面就分析过了，只不错前面传进去的//参数是true（表明是核心线程），如今的参数是false表明是非核心线程。else if (!addWorker(command, false))reject(command);

  

到目前为止咱们分析完了三种策略：1）刚开始的线程先用核心线程处理 2）核心线程在忙，那么就把当前线程压到队列 3）队列满了，那么就启动

非核心线程。不过须要咱们注意的是咱们是以FixedThreadPool线程池为例的，FixedThreadPool使用的队列是LinkedBlockingQueue,这个默认给的是

是int的最大值，换句话说咱们能够认为就是无界的，由于在当要把这个队列存满的时候内存早已经溢出了，也就是说，咱们的第三个策略压根就走不到。

前面咱们分析了第一种策略的时候，咱们顺带也分析了，核心线程获取到须要执行的task之后是如何执行task的，接下来咱们须要分析。核心线程把当前任务已经处理完了，是如何从队列里面获取task任务进行处理的。

入口是从addWorker这个方法进去，以前咱们这个方法只是分析到启动了线程，可是没有分析里面的线程是如何工做的，接下来分析以前启动的线程是如何工做的。

以下代码咱们以前看到过：

boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {    w = new Worker(firstTask);    final Thread t = w.thread;    if (t != null) {        final ReentrantLock mainLock = this.mainLock;        mainLock.lock();        try {            // Recheck while holding lock.            // Back out on ThreadFactory failure or if            // shut down before lock acquired.            int rs = runStateOf(ctl.get());            if (rs < SHUTDOWN ||                (rs == SHUTDOWN && firstTask == null)) {                if (t.isAlive()) // precheck that t is startable                    throw new IllegalThreadStateException();                workers.add(w);                int s = workers.size();                if (s > largestPoolSize)                    largestPoolSize = s;                workerAdded = true;            }        } finally {            mainLock.unlock();        }        if (workerAdded) {            //核心代码           //咱们以前分析过，其实这个start方法调用的是worker的run方法。           //worker的run方法里调用的是runWork方法           //咱们回过头来再分析一下，runworker方法            t.start();            workerStarted = true;        }    }} finally {    if (! workerStarted)        addWorkerFailed(w);}​final void runWorker(Worker w) {    Thread wt = Thread.currentThread();    Runnable task = w.firstTask;    w.firstTask = null;    w.unlock(); // allow interrupts    boolean completedAbruptly = true;    try {        //核心代码        //分析一下：若是task != null         //说明当前核心线程有任务运行，而后直接运行task便可。        //若是task == null 那么说明核心线程已经把任务运行完了，         //那么由于条件是|| 因此接下        //来会走task = getTask()) != null 代码， getTask()就是从队列里面去获取任务        //而后运行的。        while (task != null || (task = getTask()) != null) {            //加锁            w.lock();            // If pool is stopping, ensure thread is interrupted;            // if not, ensure thread is not interrupted.  This            // requires a recheck in second case to deal with            // shutdownNow race while clearing interrupt            if ((runStateAtLeast(ctl.get(), STOP) ||                 (Thread.interrupted() &&                  runStateAtLeast(ctl.get(), STOP))) &&                !wt.isInterrupted())                wt.interrupt();            try {                beforeExecute(wt, task);                Throwable thrown = null;                try {                    task.run();                } catch (RuntimeException x) {                    thrown = x; throw x;                } catch (Error x) {                    thrown = x; throw x;                } catch (Throwable x) {                    thrown = x; throw new Error(x);                } finally {                    afterExecute(task, thrown);                }            } finally {                task = null;                w.completedTasks++;                //释放锁                w.unlock();            }        }        completedAbruptly = false;    } finally {        processWorkerExit(w, completedAbruptly);    }}

  接下来分析一下getTask方法，看一下是如何从队列里面获取task任务的？

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?        //allowCoreThreadTimeOut 默认是false        //当前线程数有没有超过corePoolSize 正常状况下是flase        //因此timed的值是false        //注：allowCoreThreadTimeOut=true表明着，核心线程若是空闲超过必定时间就回收        //超过核心线程的线程 空闲超过必定时间后会回收线程。        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;        if ((wc > maximumPoolSize || (timed && timedOut))           && (wc > 1 || workQueue.isEmpty())) {            if (compareAndDecrementWorkerCount(c))                return null;            continue;        }        try {            //这个地方代码比较关键            //若是timed = ture 那么执行的是 workQueue.poll方法  这个方法没有阻塞的功能            //若是timed = false那么指定是 workerQueue.take方法 若是方法带有阻塞的功能            Runnable r = timed ?                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :                //执行这个方法，从阻塞队列中获取任务。                workQueue.take();            if (r != null)                return r;            timedOut = true;        } catch (InterruptedException retry) {           timedOut = false;        }    }}

 

   到此其实咱们把FixedThreadPool的原理就搞明白了，其他的几个线程池比较简单，你们能够本身看一下。咱们画图总结一下这个线程池的原理：