ThreadPoolExecutor原理解析（一）

78 阅读 0 评论 52 点赞

我是靠谱客的博主粗心画板，最近开发中收集的这篇文章主要介绍ThreadPoolExecutor原理解析（一），觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

Java中的线程池核心实现类是ThreadPoolExecutor，本次基于JDK 1.8的源码来分析Java线程池的核心设计与实现。

ThreadPoolExecutor的UML类图：

ThreadPoolExecutor实现的顶层接口是Executor，顶层接口Executor提供了一种思想：将任务提交和任务执行进行解耦。用户无需关注如何创建线程，如何调度线程来执行任务，只需提供Runnable对象，将任务的运行逻辑提交到执行器(Executor)中，由Executor框架完成线程的调配和任务的执行部分。

ExecutorService接口增加了一些能力：（1）扩充执行任务的能力，补充可以为一个或一批异步任务生成Future的方法；2）提供了管控线程池的方法，比如停止线程池的运行。

AbstractExecutorService则是上层的抽象类，将执行任务的流程串联了起来，保证下层的实现只需关注一个执行任务的方法即可。最下层的实现类ThreadPoolExecutor实现最复杂的运行部分。

ThreadPoolExecutor的构造方法

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    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;
    }

ThreadPoolExecutor提供了四种构造方法，这里以参数最多的构造方法进行分析，构造函数的参数含义如下：

corePoolSize 核心线程数大小

当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使有其他空闲线程可以处理任务也会创建新线程，当工作的线程数大于核心线程时就不会在创建，如果调用了prestartAllCoreThreads 方法，线程池会提前把核心线程都创造好，并启动。

maximumPoolsize允许创建的最大线程数

此值必须大于等于1。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。如果使用了无界队列，那么所有的任务会加入队列，这个参数就没有什么效果了。

keepAliveTime多余的空闲线程的存活时间

当前线程池中线程数量超过 corePoolSize 时，当空闲时间达到 KeepAliveTime 时，多余线程会被销毁直到只剩下 corePoolSize 个线程为止；如果任务很多，并且每个任务的执行时间比较短，避免线程重复创建和回收，可以调大这个时间，提高线程的利用率。

unit线程存活时间单位

workQueue等待队列

– ArrayBlockingQueue:一种有界队列，按照FIFO原则进行排序，支持公平锁和非公平锁；

– LinkedBlockingQueue: 由链表组成的有界队列，FIFO，默认创建的队列为Integer.MAX_VALUE；

– PriorityBlockingQueue：支持线程优先排序的无界队列，不能保证同优先级的线程间的排序；

– DelayQueue：实现了PriorityBlockingQueue 的延时获取的无界队列；

– LinkedBlockingDeque：由链表结构组成的双向阻塞队列；

threadFactory线程工厂（主要用来添加线程名）

表示生成线程池中工作线程的线程工厂，用于创建线程，一般默认的即可。也可以通过线程工厂给每个创建出来的线程设置更有意义的名字

handler拒绝策略

– AbortPoilcy：默认的拒绝策略，当任务不能再提交时，直接抛出；

– CallerRunsPoilcy：由调用线程运行，（提交异步任务的主线程）；

– DiscardPolicy：抛弃最新添加的任务，其他什么也不做；

– DiscardOldestPoilcy：抛弃队列最前面的任务，然后加入最新的任务；

– 实现RejectedExecutionHandler接口，自己扩展RejectedExecutionHandler接口，定义自己的拒绝策略。

ThreadPoolExecutor的execute方法

    /**
     * Executes the given task sometime in the future.  The task
     * may execute in a new thread or in an existing pooled thread.
     *
     * If the task cannot be submitted for execution, either because this
     * executor has been shutdown or because its capacity has been reached,
     * the task is handled by the current {@code RejectedExecutionHandler}.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     *         {@code RejectedExecutionHandler}, if the task
     *         cannot be accepted for execution
     * @throws NullPointerException if {@code command} is null
     */
    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.
         */
        // 获取ctl，ctl初始值是ctlOf(RUNNING, 0)，表示线程池处于运行中，工作线程数为0
        int c = ctl.get();
        // 如果工作线程数小于corePoolSize，则添加工作线程
        if (workerCountOf(c) < corePoolSize) {
           // 使用入参任务通过addWord方法创建一个新的线程，
           // 如果能完成新线程创建exexute方法结束，成功提交任务
            if (addWorker(command, true))
                return;
           // 如果添加核心工作线程失败，那就重新获取ctl，可能是线程池状态被其他线程修改了
           // 也可能是其他线程也在向线程池提交任务，导致核心工作线程已经超过了corePoolSize
            c = ctl.get();
        }
        // 判断线程池状态是否还是RUNNING，如果是就把任务添加到阻塞队列中
        if (isRunning(c) && workQueue.offer(command)) {
            // 再次检查线程池的状态，如果线程池不是RUNNING了，那就不能再接受任务了。
            // 就得把任务从队列中移除，并执行拒绝策略
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            // 如果线程池的状态没有发生改变，仍然是RUNNING，那就不需要把任务从队列中移除掉
            // 这里是一个担保机制，为了确保刚刚入队的任务有线程会去处理它。
            // 需要判断一下工作线程数，如果为0，那就添加一个非核心的工作线程,
            // 添加的这个线程没有自己的任务，目的就是从队列中获取任务来执行
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
            //如果以上都不能加入任务到工作队列，将尝试使用任务新增一个线程，
            // 这里的false表示创建非核心工作线程，判断当前工作线程数是否超过了maximumPoolSize
            //如果失败，则是线程池已经shutdown或者线程池已经达到饱和状态，所以reject;
        else if (!addWorker(command, false))
            reject(command);
    }