ThreadPoolExecutor 线程池简单解析

jdk1.8 ThreadPoolExecutor

ThreadPoolExecutor实际就是java的线程池，开发常用的Executors.newxxxxx()来创建不同类型和作用的线程池，其底部实际都是ThreadPoolExecutor。

1.创建构造方法
以最多参数的为例

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;

    }

    corePoolSize：核心的线程数

    maximumPoolSize：最大线程数

    keepAliveTime：当现有线程数大于核心线程数时，没有执行任务的线程存活时间    CachedThreadPool弹性线程池有用60s，其它是0表示不回收

    unit：存活时间单位

    workQueue：阻塞的工作队列存放需要执行的任务

    threadFactory：创建线程的工厂，实际就是给每个线程一个名，常见的pool-1-thread名就是它完成的

    handler：线程池饱和策略，当工作队列满了，线程池创建的线程满了，就是忙不过来的时候线程池的处理方案

    workQueue：阻塞队列实现主要有5种

        1、ArrayBlockingQueue：数组结构的有界阻塞队列，按FIFO排序任务；

        2、LinkedBlockingQuene：链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene；

                固定线程池、单线程池

        3、SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQuene；

                弹性线程池

        4、priorityBlockingQuene：具有优先级的无界阻塞队列；

        5、DelayedWorkQueue：有定时性子的阻塞队列

                定时线程池

    handler：线程池饱和策略，主要有4种

        1、AbortPolicy：直接抛出异常，默认策略；

        2、CallerRunsPolicy：用调用者所在的线程来执行任务；

        3、DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；

        4、DiscardPolicy：直接丢弃任务；

2.线程池执行任务execute方法
常用的invokeAll、invokeAny、submit是上层ExecutorService又封装的执行策略，底层只有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.

         *

         * 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.

         *

         * 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.

         */

        int c = ctl.get();

        if (workerCountOf(c) < corePoolSize) {

            if (addWorker(command, true))

                return;

            c = ctl.get();

        }

        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);

        }

        else if (!addWorker(command, false))

            reject(command);

    }

ctl:需要详细学习，顺便都看了
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//初始高3位1后面全0,从后面代码可以知道，ctl实际是用高3位表示线程池状态，后面29位表示当前线程数
private static final int COUNT_BITS = Integer.SIZE - 3;//32-3=29
private static final int CAPACITY = (1 << COUNT_BITS) - 1;

//明显这5个代表当前线程池状态，实际使用高3位来表示
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;//-1左移位29结果11100000000000000000000000000000 高3位是1 接受新任务并且处理已经进入阻塞队列的任务
private static final int SHUTDOWN = 0 << COUNT_BITS;//000 不接受新任务，但是处理已经进入阻塞队列的任务
private static final int STOP = 1 << COUNT_BITS;//001 不接受新任务，不处理已经进入阻塞队列的任务并且中断正在运行的任务
private static final int TIDYING = 2 << COUNT_BITS;//010 所有的任务都已经终止，workerCount为0，线程转化为TIDYING状态并且调用terminated钩子函数
private static final int TERMINATED = 3 << COUNT_BITS;//011 terminated钩子函数已经运行完成

// Packing and unpacking ctl
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; }//安位或

一句一句：
int c = ctl.get(); 通过上面知道c有两个作用，线程池状态，线程数

if (workerCountOf(c) < corePoolSize) { workerCountOf(c)就不详细看了，作用是根据低29位获取当前线程数，判断小于核心线程数corePoolSize

if (addWorker(command, true)) 说明是小于核心线程数的这里就创建新线程执行该任务，创建成功那就没后面事了

c = ctl.get(); 这里说明创建新线程失败了，线程池大多是并发场景，别的线程先一步创建就可能导致超过核心线程数，这里再次获取线程池 ctl，因为可以肯定是有变化的

if (isRunning(c) && workQueue.offer(command)) { 能到这说明线程池的线程数>=corePoolSize，或者创建新线程失败也是超过核心线程数的，不管哪种情况都表明任务就需要等待了，
判断线程池是否RUNNING执行状态，是继续把当前任务添加到工作队列中入队

int recheck = ctl.get(); 再次检查，前面的检查执行状态和入队后这里的线程状态并发下是很可能有改变的

if (! isRunning(recheck) && remove(command)) 这里注意外层if检查线程池是执行状态的且入队成功的，再次检查如果变更了不是执行中，说明线程池关了，已经不再执行任务了，
当前任务从工作队列移除，执行饱和策略reject

else if (workerCountOf(recheck) == 0) 这里说明还是执行中，特别注意从工作队列中移除是不会失败的因为之前是保证入队成功了的。检查当前线程数没有线程就创建一个线程，
注意只是创建线程不是直接执行该任务，后续会取队列执行

else if (!addWorker(command, false)) 说明线程池非执行状态或提交任务失败，则创建新线程执行任务，实际是尝试提高线程数到最大线程数，addWorker中会判断线程池状态和线程数的，
失败就饱和策略

3.addworker创建新线程执行任务

private boolean addWorker(Runnable firstTask, boolean core) {

        retry: //标签

        for (;;) { //死循环 jdk多线程常干的事，然后CAS

            int c = ctl.get(); //获取线程状态

            int rs = runStateOf(c);//获取高3位即线程池状态

            // Check if queue empty only if necessary.

            if (rs >= SHUTDOWN && //线程池非运行状态

                ! (rs == SHUTDOWN && // 状态为SHUTDOWN

                   firstTask == null && // 传递的任务为null，对于execute方法线程池满编入队之后 会在这里过滤掉执行后面创建新线程去执行队列

                   ! workQueue.isEmpty())) // worker队列不为空   这里也说明SHUTDOWN还会继续处理队列中的任务

                return false;

            for (;;) {

                int wc = workerCountOf(c);    //当前线程数量

                if (wc >= CAPACITY ||    //线程数大于等于最大容量

                    wc >= (core ? corePoolSize : maximumPoolSize)) //线程数大于等于核心线程数或最大线程数  根据传值判断

                    return false;    //直接结束

                if (compareAndIncrementWorkerCount(c))    // CAS增加线程数的数量

                    break retry; //跳出外层循环

                c = ctl.get();  // Re-read ctl CAS增加线程数数量失败，再次获取线程池状态

                if (runStateOf(c) != rs) // 状态如果变更则说明需要重新获取状态，即跳出内层循环再次执行外循环，没变更继续执行内层循环

                    continue retry;

                // else CAS failed due to workerCount change; retry inner loop

            }

        }

        //实际上面这部分就是CAS增加线程数，保证成功

        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);    //加锁的用意在这，保证并发下hashset插入正确

                        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;

    }

4.核心线程Worker相关

private final class Worker

        extends AbstractQueuedSynchronizer

        implements Runnable

    AQS模式 独占模式的AQS 实际也是线程池的线程调度实现方式

    /**

     * Creates with given first task and thread from ThreadFactory.

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

     */

    Worker(Runnable firstTask) {

        setState(-1); // inhibit interrupts until runWorker    设置AQS本节点状态是正常调度状态

        this.firstTask = firstTask;

        this.thread = getThreadFactory().newThread(this);//用给定的或默认的线程工厂创建新线程

    }

    /** Delegates main run loop to outer runWorker  */

    public void run() {

        runWorker(this);//执行线程start 实际是执行这

    }

5.runWorker

final void runWorker(Worker w) {

        Thread wt = Thread.currentThread();

        Runnable task = w.firstTask;

        w.firstTask = null;

        w.unlock(); // allow interrupts 这一步解锁，看AQS知道1初始化本节点状态0，2唤醒AQS的线程调度

        boolean completedAbruptly = true;

        try {

            while (task != null || (task = getTask()) != null) {//获取任务去执行

                w.lock();//实际是把本线程放入AQS调度队列中等待调度

                // 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;//标识是否是异常完成,上面是死循环,没任务了线程中断肯定是false，因异常中断的就是true

        } finally {

            processWorkerExit(w, completedAbruptly);//主要作用是没任务了,检查是否需要中断本线程

        }

    }

总结：
线程池实际就是线程安全的用hashset维护一个指定策略的线程集，用于执行提交的任务。其中使用AQS独占模式线程调度

巴特西

ThreadPoolExecutor 线程池简单解析

最新文章

热门文章

巴特西

ThreadPoolExecutor 线程池 简单解析

最新文章

热门文章

ThreadPoolExecutor 线程池简单解析