ThreadPoolExecutor原理解析（一）

青旅半醒 2024-03-26 12:38 30阅读 0赞

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

### ThreadPoolExecutor的UML类图： ###

![41d06cea4f06882f3dc4119a854225f2.png][]

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

### ThreadPoolExecutor.execute方法运行流程 ###

![a0b3f186690ecbcd677c1a27226268a0.png][]

[41d06cea4f06882f3dc4119a854225f2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/569e2948537b43509faefb02443c2376.png
[a0b3f186690ecbcd677c1a27226268a0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/1ae713f2ffd54540bba3c26f4f82ab80.png