ThreadPoolExecutor线程池是如何实现的？

本文基于 JDK1.8 编写。

1. 背景

当只有一个任务时，使用单个线程执行可以很好地满足需求；但由于单个线程是串行执行任务的，随着任务的增多，处理速度会很慢，如下代码：

List<Task> tasks = IntStream.range(1, 100)
                .mapToObj(Task::new)
                .collect(Collectors.toList());
tasks.forEach(Task::run);

为了提高多个任务的处理速度，可以使用多个线程执行任务，如下代码：

tasks.stream().map(Thread::new).forEach(Thread::start);

上面的代码为每个要执行的任务都创建了一个线程，但实际却不能这样做，有以下几个方面的原因：

• 线程的创建和销毁需要消耗资源，并不是零成本的。
• CPU的处理速度是有限的，创建过多的线程并不能提高速度。
• 受限于操作系统和JVM栈的限制，并不能创建无限多的线程。

可以使用线程池解决以上问题，线程池采用池化的思想，将线程视为可以共享的资源以实现线程的复用，如下代码：

ExecutorService executorService = Executors.newFixedThreadPool(10);
tasks.forEach(executorService::execute);

在JDK中线程池的实现是ThreadPoolExecutor，下面让我们看下ThreadPoolExecutor是如何设计的？

2. ThreadPoolExecutor是如何设计的？

2.1. 线程池有哪些核心参数？

ThreadPoolExecutor线程池有以下几个核心参数：

• corePoolSize：核心线程数。
• maximumPoolSize：最大线程数。
• keepAliveTime和unit：非核心线程的空闲存活时间及单位。
• workQueue：阻塞队列。
• threadFactory：线程工厂。
• handler：拒绝策略。

以上参数决定了线程池的行为，在线程池处理任务时会使用到。

2.2. 线程池是如何处理任务的？

当一个任务被提交到ThreadPoolExecutor线程池时，会首先判断线程池中的线程数是否达到核心线程数corePoolSize：

• 如果没有达到，则使用线程工厂threadFactory创建新的线程执行该任务并返回。

• 如果达到，则将该任务加入到阻塞队列workQueue中，等待执行。

如果阻塞队列已经满了，则判断线程池中的线程数是否达到最大线程数maximumPoolSize：

• 如果没有达到，则使用线程工厂threadFactory创建新的线程执行该任务并返回。
• 如果达到，则执行拒绝策略handler。

另外，当线程的空闲时间超过keepAliveTime且线程池中的线程数大于核心线程数时，会销毁线程。

类比发廊中的理发师和顾客：

起初每来一个顾客时，都会分配一个理发师；当没有多余的理发师时，顾客只能在排队等待理发师空闲；当排队等待的顾客太多时，会招来了几个理发师，来缓解发廊的压力。

当所有的理发师都在忙碌且排队等待的顾客还是太多时，只能拒绝新的顾客。当顾客没有这么多且某个理发师很久都没有工作时，理发师就可以被解雇了（理论上）。现实中为什么没有这样的发廊？原因是招聘解雇理发师的成本要大于顾客带来的消费。

2.3. 线程池有哪些拒绝策略？

ThreadPoolExecutor线程池提供了四种拒绝策略RejectedExecutionHandler：

• CallerRunsPolicy：调用运行策略，会在调用execute方法的线程中执行被拒绝的任务。
• AbortPolicy：终止策略，会抛出拒绝执行异常RejectedExecutionException，也是默认的拒绝策略。
• DiscardPolicy：放弃策略，空操作。
• DiscardOldestPolicy：放弃最旧策略，会使阻塞队列中最旧的任务出队，并重新提交当前任务。

除了上面的拒绝策略，还可以自定义拒绝策略，如下代码是日志拒绝策略装饰器：

public class LoggerDecoratorPolicy implements RejectedExecutionHandler {
    private final RejectedExecutionHandler rejectionHandler;
    
    private LoggerDecoratorPolicy(RejectedExecutionHandler rejectionHandler) {
        this.rejectionHandler = rejectionHandler;
    }
    
    public static RejectedExecutionHandler decorate(RejectedExecutionHandler rejectionHandler) {
        return new LoggerDecoratorPolicy(rejectionHandler);
    }
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("[" + executor + "] Rejected task " + r);
        rejectionHandler.rejectedExecution(r, executor);
    }
}

2.4. JDK中有哪些阻塞队列？

JDK的并发包中提供了以下几种阻塞队列：

• ArrayBlockingQueue：基于数组的有界阻塞队列。
• SynchronousQueue：可以看作大小为0的阻塞队列。
• LinkedBlockingQueue：基于链表的有界阻塞队列。
• PriorityBlockingQueue：基于优先级的无界阻塞队列。

3. 如何使用线程池？

3.1. 常用的线程池有哪些？

在JDK的线程池工厂Executors中提供了以下四种常用的线程池：

• newSingleThreadExecutor：单线程的线程池。
• newCachedThreadPool：缓存的线程池。
• newFixedThreadPool：固定大小的线程池。
• newScheduledThreadPool：定时和周期性执行的线程池。

在线程池的核心参数设置上，这些线程池有以下不同：

3.2. 如何自定义线程池？

如果线程池工厂Executors提供的线程池不能满足需求，可以考虑通过自定义线程池的方式，即根据需求设置线程池的核心参数。如下示例：

ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                8,
                8,
                0L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                Executors.defaultThreadFactory(),
                LoggerDecoratorPolicy.decorate(new ThreadPoolExecutor.AbortPolicy()));

3.3. 如何设置合适的线程数？

• 对于CPU密集型任务，线程数建议设置为CPU核心数 + 1^[1]。
• 对于IO密集型任务，线程数建议设置为CPU核心数 * (1 + IO耗时 / CPU耗时)。

3.4. 如何提交和终止任务？

先线程池中提交任务有以下两种方式：

• Executor#execute：提交Runnable类型的任务，并且没有返回值。
• AbstractExecutorService#submit(Callable<T>)：提交Callable类型的任务，并且返回Future。

终止线程池中的任务有以下两种方式：

• ThreadPoolExecutor#shutdown：关闭线程池，不接收新的任务但会将已提交的任务处理完。
• ThreadPoolExecutor#shutdownNow：立即关闭线程池，中断执行中的任务，返回等待的任务列表。

4. ThreadPoolExecutor源码浅析

介于笔者水平有限，这里仅展示关键节点，不逐行解析。

4.1. 构造器

public ThreadPoolExecutor(int corePoolSize, // 核心线程数
                          int maximumPoolSize, // 最大线程数
                          long keepAliveTime, // 空闲存活时间
                          TimeUnit unit, // 空闲存活时间的单位
                          BlockingQueue<Runnable> workQueue, // 阻塞队列
                          ThreadFactory threadFactory, // 线程工厂
                          RejectedExecutionHandler handler) // 拒绝策略

4.2. 提交任务

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

4.3. 创建worker

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;

        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            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
        }
    }

    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        // 创建新的 Worker
        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);
                    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.4. Worker运行逻辑

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 找到 Worker 关联的任务
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 如果有任务则执行，没有任务则等待从阻塞队列中获取
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // 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;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

4.5. 任务的获取

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

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

        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }

        int wc = workerCountOf(c);

        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

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

5. 参考文档