软件开发定制定制Java线程池ThreadPoolExecutor详解（一篇就够了）

前言

软件开发定制定制创建一个新的线程可以软件开发定制定制通过继承或者实现Runnable软件开发定制定制接口来实现，软件开发定制定制这两种方式创建的线程软件开发定制定制在运行结束后会被虚拟机销毁，软件开发定制定制进行垃圾回收，软件开发定制定制如果线程数量过多，软件开发定制定制频繁的创建和销毁线程软件开发定制定制会浪费资源，降低效率。软件开发定制定制而线程池的引入就很好软件开发定制定制解决了上述问题，软件开发定制定制线程池可以更好的创建、维护、软件开发定制定制管理线程的生命周期，做到复用，软件开发定制定制提高资源的使用效率，软件开发定制定制也避免了开发人员滥用new软件开发定制定制关键字创建线程的不规范行为。

说明：软件开发定制定制阿里开发手册中明确指出，软件开发定制定制在实际生产中，软件开发定制定制线程资源必须通过提供，软件开发定制定制不允许在应用中显式的创建线程。软件开发定制定制如果不使用线程池，软件开发定制定制有可能造成系统创建大软件开发定制定制量同类线程而导致消耗软件开发定制定制完内存或者“过度切换”的问题。

软件开发定制定制接下来主要对Java软件开发定制定制中线程池核心实现类ThreadPoolExecutor软件开发定制定制核心参数及工作原理、Executors工具类等，进行说明。

ThreadPoolExecutor

ThreadPoolExecutor软件开发定制定制是线程池的核心实现类，在JDK1.5引入，位于java.util.concurrent包，由Doug Lea完成。

Executor接口

Executor软件开发定制定制是线程池的顶层接口，JDK1.5软件开发定制定制开始引入了，位于java.util.concurrent 包。

public interface Executor {
    // 软件开发定制定制该接口中只定义了一个Runnable作为入参的execute方法
    void execute(Runnable command);
}

查看Executor接口的实现类图

• Executor线程池相关顶级接口，它将任务的提交与任务的执行分离开来
• ExecutorService继承并扩展了Executor接口，提供了Runnable、FutureTask等主要线程实现接口扩展
• ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务
• ScheduledExecutorService继承ExecutorService接口，并定义延迟或定期执行的方法
• ScheduledThreadPoolExecutor继承ThreadPoolExecutor并实现了ScheduledExecutorService接口，是延时执行类任务的主要实现

线程存在生命周期，同样线程池也有生命周期，源码中定义了五种状态。

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;
private static final int TERMINATED =  3 << COUNT_BITS;

关于线程池状态间转换如下图所示：

 

构造方法

如何利用ThreadPoolExecutor创建一个线程池，查看其构造方法

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包含了7个核心参数，参数含义：

• corePoolSize：核心线程池的大小
• maximumPoolSize：最大线程池的大小
• keepAliveTime：当线程池中线程数大于corePoolSize，并且没有可执行任务时大于corePoolSize那部分线程的存活时间
• unit：keepAliveTime的时间单位
• workQueue：用来暂时保存任务的工作队列
• threadFactory：线程工厂提供线程的创建方式，默认使用Executors.defaultThreadFactory()
• handler：当线程池所处理的任务数超过其承载容量或关闭后继续有任务提交时，所调用的拒绝策略

核心参数

ThreadPoolExecutor中包含了七大核心参数，如果需要对线程池进行定制化操作，需要对其中比较核心的参数进行一定程度的认识。

corePoolSize

ThreadPoolExecutor会根据corePoolSize和maximumPoolSize在构造方法中设置的边界值自动调整池大小，也可以使用setCorePoolSize和setMaximumPoolSize动态更改，关于线程数量的自动调整分为以下两种场景：

• 线程数量小于corePoolSize

当在线程池中提交了一个新任务，并且运行的线程少于corePoolSize时，即使其他工作线程处于空闲状态，也会创建一个新线程来处理该请求。

• 线程数量介于corePoolSize和maximumPoolSize之间

如果运行的线程数多于corePoolSize但少于maximumPoolSize，则仅当队列已满时才会创建新线程。

如果corePoolSize和maximumPoolSize相同，那么可以创建一个固定大小的线程池。如果maximumPoolSize被设置为无界值(Integer.MAX_VALUE)，在资源允许的前提下，意味着线程池允许容纳任意数量的并发任务。
默认情况下，即使是核心线程也会在新任务到达时开始创建和启动，如果使用非空队列创建线程池池，可以通过重写prestartCoreThread或prestartAllCoreThreads方法动态覆盖，进行线程预启动。

在实际开发中，如果需要自定义线程数量，可以参考以下公式：

其中参数含义如下：

• Ncpu是处理器的核数目，可以通过Runtime.getRuntime().availableProcessors()获得
• Ucpu是期望的CPU利用率，介于0-1之间
• W/C是等待时间与计算时间的比率

keepAliveTime

keepAliveTime参数用来来设置空闲时间。如果池当前有多个corePoolSize线程，多余的线程如果空闲时间超过将会被终止，这种机制减少了在任务数量较少时线程池资源消耗。如果某个时间需要处理的任务数量增加，则将构造新线程。使用方法setKeepAliveTime可以动态更改参数值。
默认情况下，keep-alive策略仅适用于超过corePoolSize线程的情况，但是方法allowCoreThreadTimeOut也可用于将此超时策略应用于核心线程，只要 keepAliveTime值不为零即可。

workQueue

workQueue参数用来指定存放提交任务的队列，任何BlockingQueue都可以用来传输和保存提交的任务。关于队列大小与线程数量之间存在这样的关系：

• 如果线程数少于corePoolSize，对于提交的新任务会创建一个新的线程处理，并不会把任务放入队列
• 如果线程数介于corePoolSize和maximumPoolSize之间，新提交的任务会被放入阻塞队列中
• 如果线程池处于饱和状态，即无法创建线程也无法存放在阻塞队列，那么新任务将交由拒绝策略来处理

线程池中的常用阻塞队列一般包含SynchronousQueue、LinkedBlockingQueue、ArrayBlockingQueue几种，它们都是BlockingQueue的实现类，下面进行简单介绍。

SynchronousQueue

SynchronousQueue并不能算得上一个真正的队列，虽然实现了BlockingQueue接口，但是并没有容量，不能存储任务。只是维护一组线程，在等待着把元素加入或移出队列，相当于直接交接任务给具体执行的线程。
如果没有立即可用的线程来运行任务，则尝试将任务排队失败，因此将构造一个新线程。在处理可能具有内部依赖关系的请求集时，此策略可避免锁定。这种队列方式通常需要无限的maximumPoolSizes以避免拒绝新提交的任务。当任务提交的平均到达速度快于线程处理速度时，线程存在无限增长的可能性，而CachedThreadPool正式采用这种形式。

LinkedBlockingQueue

LinkedBlockingQueue是采用链表实现的无界队列，如果使用没有预定义容量的LinkedBlockingQueue，当所有corePoolSize线程都在处理任务时，将导致新任务都会在队列中等待，不会创建超过corePoolSize个线程。这种场景下maximumPoolSize的值对于线程数量没有任何影响。
这种依托队列处理任务的方式恰与SynchronousQueue依托线程处理任务的方式相反。

ArrayBlockingQueue

ArrayBlockingQueue是通过数组实现的有界队列。有界队列在与有限的maximumPoolSizes一起使用时有助于防止资源耗尽，但可能更难以调整和控制。使用ArrayBlockingQueue可以根据应用场景，预先估计池和队列的容量，互相权衡队列大小和最大池大小：

• 使用大队列和小池：减少线程数量，可以最大限度地减少CPU使用率、操作系统资源和上下文切换开销，但可能会导致吞吐量降低
• 使用小队列大池：较大数量的线程，如果任务提交速度过快，会在短时间内提升CPU使用率，理论上可以提高系统的吞吐量。如果任务经常阻塞(如受到IO限制)，会使得CPU切换更加频繁，可能会遇到更大的调度开销，这也会降低吞吐量

threadFactory

该参数提供了线程池中线程的创建方式，这里使用了工厂模式ThreadFactory创建新线程，默认情况下，会使用 Executors.defaultThreadFactory，它创建的线程都在同一个ThreadGroup中，并具有相同的NORM_PRIORITY优先级和非守护进程状态。
也可以根据实际场景自定义ThreadFactory，可以更改线程的名称、线程组、优先级、守护程序状态等，在自定义情况下需要注意的是如果ThreadFactory在从newThread返回null时未能创建线程，则执行程序将继续，但可能无法执行任何任务。线程应该拥有“modifyThread”RuntimePermission。如果工作线程或其他使用该池的线程不具备此权限，则服务可能会降级：配置更改可能无法及时生效，关闭池可能会一直处于可以终止但未完成的状态。

handler

如果线程池处于饱和状态，没有足够的线程数或者队列空间来处理提交的任务，或者是线程池已经处于关闭状态但还在处理进行中的任务，那么继续提交的任务就会根据线程池的拒绝策略处理。
无论哪种情况，execute方法都会调用其RejectedExecutionHandler的rejectedExecution方法。线程池中提供了四个预定义的处理程序策略：

• ThreadPoolExecutor.AbortPolicy (默认)
• ThreadPoolExecutor.DiscardPolicy
• ThreadPoolExecutor.DiscardOldestPolicy
• ThreadPoolExecutor.CallerRunsPolicy

这些预定义策略都实现了RejectedExecutionHandler接口，也可以定义实现类重写拒绝策略。

AbortPolicy

查看AbortPolicy的源码，处理程序在拒绝时抛出运行时异常RejectedExecutionException 。

public static class AbortPolicy implements RejectedExecutionHandler {
    /**
     * Creates an {@code AbortPolicy}.
     */
    public AbortPolicy() { }
 
    /**
     * Always throws RejectedExecutionException.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     * @throws RejectedExecutionException always
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                " rejected from " +
                e.toString());
    }
}

DiscardPolicy

查看源码，无法执行的任务被简单地丢弃，不做任何处理。

public static class DiscardPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code DiscardPolicy}.
     */
    public DiscardPolicy() { }
 
    /**
     * Does nothing, which has the effect of discarding task r.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
}

DiscardOldestPolicy

查看源码，如果executor没有关闭，工作队列头部的任务就会被丢弃，然后重试执行(可能会再次失败，导致这个重复。

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code DiscardOldestPolicy} for the given executor.
     */
    public DiscardOldestPolicy() { }
 
    /**
     * Obtains and ignores the next task that the executor
     * would otherwise execute, if one is immediately available,
     * and then retries execution of task r, unless the executor
     * is shut down, in which case task r is instead discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            e.getQueue().poll();
            e.execute(r);
        }
    }
}

CallerRunsPolicy

查看源码，这种策略会调用执行自身的线程运行任务，这也提供了一个简单的反馈控制机制，可以减慢提交新任务的速度。

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    /**
     * Creates a {@code CallerRunsPolicy}.
     */
    public CallerRunsPolicy() { }
 
    /**
     * Executes task r in the caller's thread, unless the executor
     * has been shut down, in which case the task is discarded.
     *
     * @param r the runnable task requested to be executed
     * @param e the executor attempting to execute this task
     */
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        if (!e.isShutdown()) {
            r.run();
        }
    }
}

钩子函数

ThreadPoolExecutor提供受保护的可重写的钩子函数，用于在线程池中线程在初始化或者执行完任务后做一些特殊处理，同样也提供了在线程池终止时可以覆写的terminated方法。

beforeExecute

线程中执行Runnable之前调用的方法。此方法由将执行任务r的线程t调用，可用于重新初始化 ThreadLocals，或执行日志记录。这实现什么都不做，但可以在子类中定制。需要注意的事是，要正确嵌套多个覆盖，子类通常应在此方法的末尾调用 super.beforeExecute。查看源码：

protected void beforeExecute(Thread t, Runnable r) { }

afterExecute

在完成给定的Runnable任务时调用的方法，此方法由执行任务的线程调用。需要注意，要正确嵌套多个覆盖，子类通常应在此方法的开头调用 super.afterExecute。查看源码：

protected void afterExecute(Runnable r, Throwable t) { }

terminated

Executor终止时调用的方法。需要注意的是子类通常应在此方法中调用 super.terminated。查看源码：

protected void terminated() { }

核心源码分析

线程池的整体工作流程可以参考如下图所示：

ThreadPoolExecutor执行execute方法分下面4种情况。

1. 如果当前运行的线程少于corePoolSize,则创建新线程来执行任务(注意,执行这一步骤需要获取全局锁)。
2. 如果运行的线程等于或多于corePoolSize,则将任务加入BlockingQueue。
3. 如果无法将任务加入BlockingQueue(队列已满),则创建新的线程来处理任务(注意,执行这一步骤需要获取全局锁)。
4. 如果创建新线程将使当前运行的线程超出maximumPoolSize,任务将被拒绝,并调用RejectedExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor采取上述步骤的总体设计思路,是为了在执行execute()方法时,尽可能地避免获取全局锁(那将会是一个严重的可伸缩瓶颈)。在ThreadPoolExecutor完成预热之后(当前运行的线程数大于等于corePoolSize),几乎所有的execute()方法调用都是执行步骤2,而步骤2不需要获取全局锁。

任务单元Worker

ThreadPoolExecutor中核心任务单元是由一个Worker内部类来实现，Worker类中定义了两个重要方法runWorker方法和addWorker方法。

private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
    /**
     * This class will never be serialized, but we provide a
     * serialVersionUID to suppress a javac warning.
     */
    private static final long serialVersionUID = 6138294804551838833L;
 
    /** Thread this worker is running in.  Null if factory fails. */
    final Thread thread;
    /** Initial task to run.  Possibly null. */
    Runnable firstTask;
    /** Per-thread task counter */
    volatile long completedTasks;
 
    /**
     * 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
        this.firstTask = firstTask;
        // 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker
        this.thread = getThreadFactory().newThread(this);
    }
 
    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }
    // 省略代码...
}

addWorker和runWorker

• addWorker用来实例化任务单元Worker对象

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    // 外层自旋
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
 
        // 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价
        // (rs > SHUTDOWN) || 
        // (rs == SHUTDOWN && firstTask != null) || 
        // (rs == SHUTDOWN && workQueue.isEmpty())
        // 1. 线程池状态大于SHUTDOWN时，直接返回false
        // 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false
        // 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
 
        // 内层自旋
        for (;;) {
            int wc = workerCountOf(c);
            // worker数量超过容量，直接返回false
            if (wc >= CAPACITY ||
                wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 使用CAS的方式增加worker数量。
            // 若增加成功，则直接跳出外层循环进入到第二部分
            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 {
        w = new Worker(firstTask);
        final Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            // worker的添加必须是串行的，因此需要加锁
            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)) {
                    // worker已经调用过了start()方法，则不再创建worker
                    if (t.isAlive()) // precheck that t is startable
                        throw new IllegalThreadStateException();
                    // worker创建并添加到workers成功
                    workers.add(w);
                    // 更新`largestPoolSize`变量
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            // 启动worker线程
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        // worker线程启动失败，说明线程池状态发生了变化(关闭操作被执行)，需要进行shutdown相关操作
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

• runWorker是核心线程执行逻辑

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    // 调用unlock()是为了让外部可以中断
    w.unlock(); // allow interrupts
    // 这个变量用于判断是否进入过自旋(while循环)
    boolean completedAbruptly = true;
    try {
        // 这儿是自旋
        // 1. 如果firstTask不为null，则执行firstTask；
        // 2. 如果firstTask为null，则调用getTask()从队列获取任务。
        // 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
        while (task != null || (task = getTask()) != null) {
            // 这儿对worker进行加锁，是为了达到下面的目的
            // 1. 降低锁范围，提升性能
            // 2. 保证每个worker执行的任务是串行的
            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();
            // 执行任务，且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
            // 这两个方法在当前类里面为空实现。
            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 {
                // 帮助gc
                task = null;
                // 已完成任务数加一 
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 自旋操作被退出，说明线程池正在结束
        processWorkerExit(w, completedAbruptly);
    }
}

submit和execute

ThreadPoolExecutor执行任务有submit和execute两种方法，这两种方法区别在于

• submit方法有返回值，便于异常处理
• execute方法没有返回值

下面来简单介绍一下submit和execute的用法

• submit方法有三种传入参数的形式

<T> Future<T> submit(Callable<T> callable);
<T> Future<T> submit(Runnable var1, T result);
Future<?> submit(Runnable runnable);

在ExecutorService接口中定义submit方法，抽象类AbstractExecutorService实现了ExecutorService中的submit方法。

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}
 
/**
  * @throws RejectedExecutionException {@inheritDoc}
  * @throws NullPointerException       {@inheritDoc}
  */
public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}
 
/**
  * @throws RejectedExecutionException {@inheritDoc}
  * @throws NullPointerException       {@inheritDoc}
  */
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

当submit方法传入Runnable对象调用Future对象的get方法返回值为null，传入Callable对象时返回get自定义 的值，在返回结果之前，主线程会阻塞等待结果返回再执行。

class RunnableDemo implements Runnable{
    @Override
    public void run() {
        System.out.println("RunableDemo is execute");
    }
}
class CallableDemo implements Callable<String>{
    @Override
    public String call() throws Exception {
        return "Call is Done";
    }
}
public class Test {
    public static void main(String[] args) throws Exception{
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future<?> call = executorService.submit(new CallableDemo());
        System.out.println("Callable'S Result："+call.get());
        Future<?> run = executorService.submit(new RunnableDemo());
        System.out.println("Runnable'S Result："+run.get());
        System.out.println("Current Thread："+Thread.currentThread().getName());
    }
}

输出结果

Callable'S Result：Call is Done
RunableDemo is execute
Runnable'S Result：null
Current Thread：main

• execute方法只有一种，传入实现Runnable接口的对象。

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();
    // worker数量比核心线程数小，直接创建worker执行任务
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    // worker数量超过核心线程数，任务直接进入队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。
        // 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。
        if (! isRunning(recheck) && remove(command))
            reject(command);
        // 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。
    // 这儿有3点需要注意：
    // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态
    // 2. addWorker第2个参数表示是否创建核心线程
    // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作
    else if (!addWorker(command, false))
        reject(command);
}

使用submit方法可以对task执行的结果成功，失败，或者执行过程中抛出的异常及时处理，暂停处理其他task，使用execute不能及时处理程序在运行中出现的异常情况。

class CallableDemo implements Callable<String>{
    @Override
    public String call() throws Exception {
        return "Call is Done";
    }
}
public class Test {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        Future<?> call = executorService.submit(new CallableDemo());
        try {
            call.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

我们可以根据具体业务场景考虑可能出现的异常，由实现Callable的接口throws，然后由ThreadPoolExecutor调用者来处理，提高多线程场景下的容错率。

Executors类

Executors是Executor框架的工具类，提供了几种线程池创建方法，以及线程池中默认配置(如线程工厂)的处理，接下来对其中常用的几种创建线程池的方式进行说明。

newSingleThreadExecutor

** **SingleThreadExecutor使用Executors.newSingleThreadExecutor()创建，查看源码

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

其中：

• corePoolSize和maximumPoolSize都是1
• LinkedBlockingQueue是一个最大值为Integer.MAX_VALUE的无界队列
• 当线程正在执行任务，新任务会被加入到LinkedBlockingQueue队列中，任务加入队列的速度远大于核心线程处理的能力时，无界队列会一直增大到最大值，可能导致OOM

因此newSingleThreadExecutor可用于处理任务量不多，但又不想频繁的创建、销毁需要与同周期的场景。

newFixedThreadPool

FixedThreadPool使用Executors.newFixedThreadPool() 创建，查看源码

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

其中：

• corePoolSize和maximumPoolSize值均为nTreads(由入参确定)
• 存活时间为0L，超过核心线程数的空闲线程会被立即销毁
• 队列依然为LinkedBlockingQueue，当线程数达到corePoolSize时，新任务会一直在无界队列中等待
• 线程池中的线程数不会超过corePoolSize，新建任务也会一直被加入到队列等待，不会执行拒绝策略
• ThreadPoolExecutor中的7个参数，maximumPoolSize，keepAliveTime，RejectedExecutionHandler为无效参数

FixedThreadPool同SingleThreadExecutor，如果nThreads的值设置过小，在任务量过多的场景下，会有可能由于线程数过少，导致任务一直堆积在队列中而引发OOM，相对好处是可以多线程处理，一定程度提高处理效率。

newCachedThreadPool

CachedThreadPool使用Executors.newCachedThreadPool()创建，查看源码

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

其中：

• corePoolSize为0，maximumPoolSize为Integer.MAX_VALUE，即maximumPool是无界的
• keepAliveTime为60L，空闲线程等待新任务的最长时间为60秒，超过60秒后将会被终止
• SynchronousQueue为线程池的工作队列，没有容量，不能存放任务

CachedThreadPool中，如果主线程提交任务的速度高于maximumPool中线程处理任务的速度时，会不断创建新线程，最终导致创建过多线程而耗尽CPU和内存资源。忽略CPU和内存消耗，某种程度上CachedThreadPool可以快速解决短时间并发问题，由于核心线程数为0，并且设置了存活时间，这些临时开辟的线程会在任务处理后被回收掉。
Executors是为了屏蔽线程池过多的参数设置影响易用性而提供的，而实际上这些参数恰恰也不需要进行逐个设置，Executors也是源码中推荐的一种使用方式，但是需要熟知他们各自的特点。

总结

Executor框架主要由三部分组成，任务，任务的执行者，执行结果，ThreadPoolExecutor和ScheduledThreadPoolExecutor的设计思想也是将这三个关键要素进行了解耦，将任务的提交和执行分离。

• 任务
  在ThreadPoolExecutor和ScheduledThreadPoolExecutor中任务是指实现了Runnable接口和Callable接口的类，ThreadPoolExecutor中将任务转换成FutureTask类，ScheduledThreadPoolExecutor中任务被转换成ScheduledFutureTask类，该类继承FutureTask，并重写了run方法，实现了延时执行任务和周期性执行任务。
• 任务的执行者
  包括任务执行机制的核心接口Executor，以及继承自Executor的ExecutorService接口和两个关键类(实现了ExecutorService接口的ThreadPoolExecutor和ScheduledThreadPoolExecutor类)。任务的执行机制，交由Worker类，进一步封装了Thread向线程池提交任务，ThreadPoolExecutor的execute方法和submit方法，以及ScheduledThreadPoolExecutor的schedule方法都是先将任务移到阻塞队列中，然后通过addWorker方法新建Worker对象，并通过runWorker方法启动线程，不断的从阻塞对列中获取异步任务交给Worker执行，直至阻塞队列中任务执行完为止。
• 执行结果
  包括接口Future和实现Future接口的FutureTask类，获取任务执行结果，在ThreadPoolExecutor中提交任务后实际上为FutureTask类，在ScheduledThreadPoolExecutor中则是ScheduledFutureTask类。

参考资料：Java并发编程的艺术并发编程网：