这是我参与更文挑战的第9天，活动详情查看：更文挑战markdown

execute()原理

执行步骤以下：oop

若是workerCount小于corePoolSize，将任务做为first Task新建线程来执行任务。将任务添加到队列中，若是添加成功依然须要检查，在进入任务以前，post

若是线程池被关闭，那么将任务从队列中移除；ui

若是当前线程池中没有工做线程，而刚刚在队列中加入了任务，要保证线程池中至少有一个工做线程能够处理任务。this

若是不能将任务加入队列中，尝试新加一个线程来执行任务，可是并不定会成功，多是线程池被shut down或者线程池达到了饱和（maximumPoolSize），若是失败了执行拒绝策略。spa

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        
        int c = ctl.get();
        // 高3位表示状态，低29位任务数量。
　　　　//工做线程小于核心线程数，建立新的线程。
        if (workerCountOf(c) < corePoolSize) {
        //建立新的worker当即执行command，轮询workQueue处理task。
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //线程池在运行状态且能够将task插入队列
　　　　//第一次校验线程池在运行状态
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //第二次校验，防止在第一次校验经过后线程池关闭。若是线程池关闭，在队列中删除task并拒绝task
            if (! isRunning(recheck) && remove(command))
                reject(command);
                //若是线程数=0（线程都死掉了，好比：corePoolSize=0），新建线程且未指定firstTask，仅仅去轮训workQueue
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //线程队列已满，尝试建立新线程执行task，建立失败后拒绝task，建立失败缘由：1.线程池关闭；2.线程数已经达到maxPoolSize
        else if (!addWorker(command, false))
            reject(command);
    }
复制代码

addWorker方法

addWorker方法主要有两部分线程

一：判断是否能够建立worker。根据自旋、CAS、ctl等，判断继续建立仍是返回false，自旋周期通常很短。code

二：同步建立workder，启动线程。 addWorker会根据当前线程池的工做状态和给定的界限限制（corePoolSize 和maximumPoolSize）判断是否能够添加新的工做线程。其中每个Worker对象都是一个AQS队列。orm

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
    　　　　 //外层循环判断线程池的状态
         for (;;) {
            //在进行获取一次上下文操做机制
            int c = ctl.get();
            int rs = runStateOf(c);
             //线程池状态
            // Check if queue empty only if necessary.
    　　　　  // 线程池状态:RUNNING = -一、SHUTDOWN = 0、 STOP = 一、TIDYING = 二、TERMINATED = 3　
　　　　　　　　//线程池至少是shutdown状态
　　　　　　　 if (rs >= SHUTDOWN &&
　　　　　　　　　　// 除了线程池正在关闭（shutdown），
　　　　　　　　　　// 队列里还有未处理的task的状况，其余都不能添加
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;
            //内层循环判断是否到达容量上限，worker+1
            for (;;) {
                int wc = workerCountOf(c);//worker数量
                //worker大于Integer最大上限
                //或到达边界上限
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //CAS worker+1
                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 
　　　　　　　　　// CAS失败workerCount被其余线程改变，
　　　　　　　　　// 从新尝试内层循环CAS对workerCount+1

            }
        }
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            final ReentrantLock mainLock = this.mainLock;
            w = new Worker(firstTask); 
            //1.state置为-1，Worker继承了AbstractQueuedSynchronizer.
            //2.设置firstTask属性.
            //3.Worker实现了Runable接口，将this做为入参建立线程.
            final Thread t = w.thread;
            if (t != null) {
　　　　　　　　　　//addWorker须要加锁
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int c = ctl.get();
                    int rs = runStateOf(c);
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        workers.add(w);//workers是HashSet<Worker>
　　　　　　　　　　　　　　//设置最大线程池大小
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (!workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
复制代码

Worker

Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }

复制代码

Worker方法实现了Runnable，它将构造的Thread赋值给thread。第一步中的setState(-1)，即将AQS中的同步状态设置为-1，线程池使用AQS中的同步状态来判断该工做线程是否能够被中断。-1：初始化值，此时工做线程尚未启动，也没有中断的必要；0：表示接受中断，此时工做线程为空闲状态；1：表示此时工做线程正在执行任务。它实现了非重入互斥锁，非重入是为了不线程池的一些控制方法得到重入锁。注意Worker实现锁的目的与传统锁的意义不太同样。其主要是为了控制线程是否可interrupt，以及其余的监控，如线程是否active（正在执行任务）。对象

private final class Worker extends AbstractQueuedSynchronizer implements Runnable
{
    private static final long serialVersionUID = 6138294804551838833L;

    //每一个worker有本身的内部线程，ThreadFactory建立失败时是null
    //封装任务线程机制
    final Thread thread;
    //初始化任务，多是null
    Runnable firstTask;
    //每一个worker的完成任务数
    volatile long completedTasks;

    Worker(Runnable firstTask) {
        // inhibit interrupts until runWorker //状态置为-1，若是中断线程须要CAS将state 从0-
        >1，以此来保证能只中断从workerQueue getTask的线程
        setState(-1); // 禁止线程在启动前被打断
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    //核心方法
    public void run() {
    //首先执行w.unlock,就是把state置为0，对该线程的中断就能够进行了
        runWorker(this);
    }

    // state = 0 表明未锁；state = 1 表明已锁

    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    // 在setCorePoolSize/shutdown等方法中断worker线程时须要调用该方法，
    // 确保中断的是从workerQueue getTask的线程
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
            return true;
        }
        return false;
    }

    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
        setState(0);
        return true;
    }

    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    //调用tryRelease修改state=0，LockSupport.unpark(thread) 下一个等待锁的线程
    public boolean isLocked() { return isHeldExclusively(); }
    // interrupt已启动线程
    void interruptIfStarted() {
        Thread t;
        // 初始化是 state = -1，不会被interrupt
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}
复制代码

Worker的run()方法实际调用的是runWorker(this)，启动工做线程（注意这里不是直接启动的任务），在工做线程中执行任务。工做线程在这循环，反复的从队列中获取任务并执行它们。

在运行全部任务以前，须要获取锁，来保证当任务在运行的时候不会被中断，除非线程池正在中止（Stop）总结以下： 1.Worker类主要负责运行线程状态的控制。 2.Worker继承了AQS实现了简单的获取锁和释放所的操做。来避免中断等待执行任务的线程时，中断正在运行中的线程（线程刚启动，还没开始执行任务）。 3.本身实现不可重入锁，是为了不在实现线程池控状态控制的方法，例如：setCorePoolSize的时候中断正在开始运行的线程。setCorePoolSize可能会调用interruptIdleWorkers()，该方法中会调用worker的tryLock()方法中断线程，本身实现锁能够确保工做线程启动以前不会被中断

runWorker的主要任务就是一直loop循环处理任务，没有任务就去getTask()，代码以下：

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // 容许被 interrupt
        boolean completedAbruptly = true;
        try {
            //循环获取任务直至 task = null （线程池关闭、超时等）
            // 注意这里的getTask()方法，咱们配置的阻塞队列会在这里起做用
            while (task != null || (task = getTask()) != null) {
                w.lock();  // 执行任务前上锁               
                // 若是线程池中止(处于STOP状态或者TIDYING、TERMINATED状态时)，设置当前线程处于中断状态若是不是(线程就处于RUNNING或者SHUTDOWN状态),确保线程不中断,从新检查当前线程池的状态是否大于等于STOP状态
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try {
                    //在线程运行前调用
                    beforeExecute(wt, task); 
                    Throwable thrown = null;
                    try {
                        //这里执行run方法
                        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，大概分为如下几个步骤。

判断线程池以及队列的状态，若是线程池状态在STOP以上，此时线程池不处理队列中的任务；或者线程池处于SHUTDOWN可是队列为空（SHUTDOWN再也不接受新的任务），workerCount减1，返回null,注意此时只是将变量减1，其实工做线程并无终止真正的终止在 processWorkerExit(w, completedAbruptly);中。若是经过了状态检查，判断是否要进行线程回收，若是须要workerCount数量减1，成功后返回null。根据timed（timed表示须要进行超时闲置线程回收），选择是限时等待仍是阻塞的方式从队列中获取任务。

方法依据配置的workQueue来工做，其阻塞原理与超时原理基于阻塞队列实现，再也不详述。

private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
	//循环
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        //线程线程池状态和队列是否为空
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
		//线程数量
        int wc = workerCountOf(c);

        
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        //（当前线程数是否大于最大线程数或者）
        //且（线程数大于1或者任务队列为空）
        //这里有个问题(timed && timedOut)timedOut = false，好像(timed && timedOut)一直都是false吧
        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;
        }
    }
}
复制代码

总结

ThreadPoolExecutor的核心是Worker，Worker实现了AbstractQueuedSynchronizer并继承了Runnable。aqs锁运用的极为巧妙。

浅析ThreadPoolExecutor源码

execute()原理

addWorker方法

Worker

getTask()

总结