ThreadPoolExecutor原理

ThreadPoolExecutor构造函数参数

/**
 * 使用给定的参数创建ThreadPoolExecutor.
 *
 * @param corePoolSize    核心线程池中的最大线程数
 * @param maximumPoolSize 总线程池中的最大线程数
 * @param keepAliveTime   空闲线程的存活时间
 * @param unit            keepAliveTime的单位
 * @param workQueue       任务队列, 保存已经提交但尚未被执行的任务
 * @param threadFactory   线程工厂(用于指定如果创建一个线程)
 * @param 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;
}

1、ThreadPoolExecutor在逻辑上将自身管理的线程池划分为两部分：核心线程池（大小对应为corePoolSize）、非核心线程池（大小对应为maximumPoolSize-corePoolSize）。
2、ThreadPoolExecutor中只有一种类型的线程，名叫Worker，它是ThreadPoolExecutor内部类，封装着Runnable任务和执行该任务的Thread对象，我们称它为【工作线程】，它也是ThreadPoolExecutor唯一需要进行维护的线程；
3、【核心线程池】【非核心线程池】都是逻辑上的概念，ThreadPoolExecutor在任务调度过程中会根据corePoolSize和maximumPoolSize的大小，判断如何执行任务。

线程池状态和管理

1、ThreadPoolExecutor内部定义了一个AtomicInteger变量—ctl，通过按位划分的方式，在一个变量中记录线程池状态和工作线程数量

1、低29位保存线程数
2、高3位保存线程池状态

//保存线程池状态和工作线程数:低29位: 工作线程数,高3位 : 线程池状态
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    private static final int COUNT_BITS = Integer.SIZE - 3;
    // 最大线程数: 2^29-1
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // 线程池状态
    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; //所有任务都已终止, 工作线程数为0, 线程转化为TIDYING状态并准备调用terminated方法
    private static final int TERMINATED =  3 << COUNT_BITS; //terminated方法已经执行完成

工作线程

工作线程（Worker），Worker内部类，实现了AQS框架，ThreadPoolExecutor通过一个HashSet来保存工作线程：

Worker定义

/**
 * Worker表示线程池中的一个工作线程, 可以与任务相关联.
 * 由于实现了AQS框架, 其同步状态值的定义如下:
 * -1: 初始状态
 * 0:  无锁状态
 * 1:  加锁状态
 */
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {

    /**
     * 与该Worker关联的线程.
     */
    final Thread thread;
    /**
     *初始化任务，可以为空，为空的时候则去任务队列workQueue里获取
     */
    Runnable firstTask;
    /**
     * 当前工作线程处理完成的任务数
     */
    volatile long completedTasks;

    Worker(Runnable firstTask) {
        setState(-1); // 初始的同步状态值
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

    /**
     * 执行任务
     */
    public void run() {
        runWorker(this);
    }

    /**
     * 是否加锁
     */
    protected boolean isHeldExclusively() {
        return getState() != 0;
    }

    /**
     * 尝试获取锁
     */
    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);
    }

    public boolean isLocked() {
        return isHeldExclusively();
    }

    /**
     * 中断线程(仅任务非初始状态)
     */
    void interruptIfStarted() {
        Thread t;
        if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
            } catch (SecurityException ignore) {
            }
        }
    }
}

线程池执行execute

execute代码逻辑

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        //获取线程池状态和工作线程数量
        int c = ctl.get();
        //如果工作线程数 < 核心线程数，则创建工作线程
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true)) //addWorker创建工作线程
                return;
            c = ctl.get();
        }
        //工作线程创建失败，或者，工作线程 >= 核心线程数，任务插入任务队列workQueue
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //再次检查线程池状态，如果不是运行状态，移除任务，并拒绝
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //如果工作线程为0，则创建一个不带任务的线程，线程自动去任务队列获取任务执行
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //任务插入队列失败，创建非核心工作线程，如果失败，则说明工作线程 > 总线程数量，则执行拒绝策略
        else if (!addWorker(command, false))
            reject(command);
    }

执行流程图

说明：
1、如果工作线程数小于核心线程池上限（CorePoolSize），则直接新建一个工作线程并执行任务；
2、如果工作线程数大于等于CorePoolSize，则尝试将任务加入到队列等待以后执行。如果队列已满，则在总线程池未满的情况下（CorePoolSize ≤ 工作线程数 ＜ maximumPoolSize）新建一个工作线程立即执行任务，否则执行拒绝策略。

创建工作线程addWorker

/**
     * 添加工作线程并执行任务
     *
     * @param firstTask 如果指定了该参数, 表示将立即执行该firstTask任务; 否则从工作队列中获取任务并执行
     * @param core      执行任务的工作线程归属于哪个线程池:  true-核心线程池  false-非核心线程池
     */
    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            /**
             * 这个if主要是判断哪些情况下, 线程池不再接受新任务执行, 而是直接返回.总结下, 有以下几种情况：
             * 1. 线程池状态为 STOP 或 TIDYING 或 TERMINATED: 线程池状态为上述任一一种时, 都不会再接受任务，所以直接返回
             * 2. 线程池状态≥ SHUTDOWN 且 firstTask != null: 因为当线程池状态≥ SHUTDOWN时, 不再接受新任务的提交，所以直接返回
             * 3. 线程池状态≥ SHUTDOWN 且 队列为空: 队列中已经没有任务了, 所以也就不需要执行任何任务了，可以直接返回
             */
            if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                            firstTask == null &&
                            ! workQueue.isEmpty()))
                return false;

            for (;;) {
                //获取工作线程数
                int wc = workerCountOf(c);
                /**
                 * 这个if主要是判断工作线程数是否超限, 以下任一情况属于超限, 直接返回:
                 * 1. 工作线程数超过最大工作线程数(2^29-1)
                 * 2. 工作线程数超过核心线程池上限(入参core为true, 表示归属核心线程池)
                 * 3. 工作线程数超过总线程池上限(入参core为false, 表示归属非核心线程池)
                 */
                if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //增加工作线程数
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();
                //CAS失败，自旋重新操作
                if (runStateOf(c) != rs)
                    continue retry;

            }
        }

        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 {
                    //重新检查线程池状态
                    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) {
                    //启动线程，这个很关键，因为t在new的时候是用Work作为任务的，
                    // Work实现了Runnale接口，所以t.start就是执行Work的run方法
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

特别注意：
启动线程，这个很关键，因为thread在new的时候是用Work（this）作为任务的, Work实现了Runnale接口，所以t.start就是执行Work的run方法。

工作线程的执行runWorker

final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        // 任务, 如果是null则从队列取任务
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // 允许执行线程被中断
        boolean completedAbruptly = true;
        try {
            // 当task==null时会通过getTask从队列取任务
            while (task != null || (task = getTask()) != null) {
                w.lock();

                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++;  // 完成任务数+1
                    w.unlock();
                }
            }
            //说明该工作线程自身既没有携带任务, 也没从任务队列中获取到任务
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }

获取任务方法getTask

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

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

            /**
             * 以下IF用于判断哪些情况下不允许再从队列获取任务:
             * 1. 线程池进入停止状态（STOP/TIDYING/TERMINATED）, 此时即使队列中还有任务未执行, 也不再执行
             * 2. 线程池非RUNNING状态, 且队列为空
             */
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }

            int wc = workerCountOf(c);

            /**
             * timed变量用于判断是否需要进行超时控制:
             * 对于核心线程池中的工作线程, 除非设置了allowCoreThreadTimeOut==true, 否则不会超时回收;
             * 对于非核心线程池中的工作线程, 都需要超时控制
             */
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            // 这里主要是当外部通过setMaximumPoolSize方法重新设置了最大线程数时
            // 需要回收多出的工作线程
            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时，传入的RejectedExecutionHandler对象，一共4种

1.AbortPolicy（默认），抛出一个RejectedExecutionException异常

2、DiscardPolicy，什么都不做，等任务自己被回收

3、DiscardOldestPolicy，丢弃任务队列中的最近一个任务，并执行当前任务

4、CallerRunsPolicy，以自身线程来执行任务,减缓新任务提交的速度

线程池关闭

1、shutdown方法将线程池切换到SHUTDOWN状态（如果已经停止，则不用切换），并调用interruptIdleWorkers方法中断所有空闲的工作线程，最后调用tryTerminate尝试结束线程池

2、shutdownNow方法的主要不同之处就是，它会将线程池的状态至少置为STOP，同时中断所有工作线程（无论该线程是空闲还是运行中），同时返回任务队列中的所有任务

原文地址：https://blog.51cto.com/janephp/2416396