ThreadPoolTaskExecutor线程是Spring的线程池，其底层是依据JDK线程池ThreadPoolExecutor来实现的
image.png
1.ThreadPoolTaskExecutor的基本参数：因为它是基于ThreadPoolExecutor来实现的，我们可以参考ThreadPoolExecutor的构造函数
image.png
corePoolSize：线程池维护线程最小的数量，默认为1
maxPoolSize：线程池维护线程最大数量，默认为Integer.MAX_VALUE
keepAliveSeconds：(maxPoolSize-corePoolSize)部分线程空闲最大存活时间，默认存活时间是60s
queueCapacity：阻塞任务队列的大小，默认为Integer.MAX_VALUE，默认使用LinkedBlockingQueue
allowCoreThreadTimeOut：设置为true的话，keepAliveSeconds参数设置的有效时间对corePoolSize线程也有效，默认是flase
threadFactory：：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。使用开源框架guava提供的ThreadFactoryBuilder可以快速给线程池里的线程设置有意义的名字
rejectedExecutionHandler：拒绝策略，当队列workQueue和线程池maxPoolSize都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常，有以下四种策略，当然也可以根据实际业务需求类实现RejectedExecutionHandler接口实现自己的处理策略
①AbortPolicy：丢弃任务，并且抛出RejectedExecutionException异常
②DiscardPolicy：丢弃任务，不处理，不抛出异常
③CallerRunsPolicy：只用调用者所在线程来运行任务
③DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务，并且重复该操作
2.线程池的执行流程：当提前任务到当前线程池时，先判断当前线程池中线程数量是否小于corePoolSize，如果小于创建新的线程处理请求，不管当前有没有线程闲置；如果大于等于，则将线程想放入阻塞队列workQueue中，线程池中存在空闲的线程后会去处理workQueue中任务；如果workQueue也满了，则会新建工作线程处理任务，当线程池中的线程大于最大线程数maxPoolSize时，则会用选定的拒绝策略来处理新的线程
image.png
下面是源码部分（使用execute()方法）：
注意：ThreadPoolTaskExecutor中execute()有重载方法，但最后都是调用同一方法，处理逻辑相同
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// ctl记录着runState和workerCount
int c = ctl.get();
// 使用workerCountOf取出当前线程池中存在的线程数，与核心线程比较大小
if (workerCountOf(c) < corePoolSize) {
// 当前线程小于核心线程，添加线程执行任务，第二个参数表示限制添加线程的数量是和核心线程corePoolSize还是和线程池维护最大线程数maxPoolSize比较
if (addWorker(command, true))
// 如果成功择直接返回结果
return;
// 失败则重新获取runState和workerCount的值
c = ctl.get();
}
// 如果当前线程池是运行状态并且任务添加到队列成功
if (isRunning(c) && workQueue.offer(command)) {
// 重新获取runState和workerCount的值
int recheck = ctl.get();
// 如果线程池不是运行状态并且任务已经被移除
if (! isRunning(recheck) && remove(command))
// 根据拒绝策略处理该任务
reject(command);
// 判断当前线程池的线程数是否为0
else if (workerCountOf(recheck) == 0)
//第一个参数为null，表示在线程池中创建一个线程，但不去启动，第二个参数为false，将线程池的有限线程数量的上限设置为maximumPoolSize
addWorker(null, false);
}
//再次调用addWorker方法，但第二个参数传入为false，将线程池的有限线程数量的上限设置为maximumPoolSize
else if (!addWorker(command, false))
// 如果失败拒绝该任务
reject(command);
}
从上看出其实线程池执行任务的主要方法是addWorker()方法，execute()方法只是将任务提交以及做一些判断，我们看一下addWorker()方法：
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
// 获取运行状态
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 如果re >= SHUTDOWN (线程池关闭不在接收任务，也不会处理队列中已有的任务)，并且rs == SHUTDOWN ，firstTask 为空，阻塞队列workQueue不为空
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
// 返回flase
return false;
for (;;) {
// 获取当前线程池中已有线程数
int wc = workerCountOf(c);
// wc >= 容器最大量或者 wc >= 核心线程数或最大维护线程数（根据core值判断以哪个为依据）
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
// 返回flase
return false;
// 增加workerCount（原子操作）
if (compareAndIncrementWorkerCount(c))
// 如果增加成功，则直接返回
break retry;
// 失败重新获取runState
c = ctl.get(); // Re-read ctl
// 判断当前runState与最开始rs是否相等，不相等说明状态已经被修改，返回重新执行
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 {
// 根据firstTask创建Worker开始执行任务
w = new Worker(firstTask);
// 根据worker创建一个线程
final Thread t = w.thread;
if (t != null) {
// 使用重入锁的目的是一文workers使用的是 HashSet线程不安全
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());
// rs < SHUTDOWN 说明线程池正在处于运行状态，rs =SHUTDOWN firstTask 为空，其中firstTask 为空表示只是新建线程但没有任务需要执行
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
// 判断创建的线程是否存活，存活则抛出异常
throw new IllegalThreadStateException();
// 加入worker中，workers中包含线程池中所有的工作线程，只有只有锁时才能访问
workers.add(w);
// 获取当前工作线程数
int s = workers.size();
if (s > largestPoolSize)
// 如果大于最大线程数，将最大线程数重新赋值
largestPoolSize = s;
// worker的添加工作状态改为true
workerAdded = true;
}
} finally {
// 释放锁
mainLock.unlock();
}
// 如果添加worker工作完成，启动线程，并修改线程启动状态
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
// 如果线程启动失败
if (! workerStarted)
// 将线程从workers中删除，该方法中也是用了重入锁，保证线程安全
addWorkerFailed(w);
}
// 返回线程启动状态
return workerStarted;
}
上面源码中调用start()方法启动线程，其实是调用Worker中run()方法来启动，因为Worker实现实现了Runnable接口，如下
image.png
调用Worker的run()方法，从图中可以看出本质执行的方法是runWorker()方法，源码如下：
final void runWorker(Worker w) {
// 获取当前线程
Thread wt = Thread.currentThread();
// 获取当前任务
Runnable task = w.firstTask;
// 将当前任务与值为空，防止其他线程拿到后执行
w.firstTask = null;
// 释放锁
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// task不为空，task ==getTask不为空，一直从线程池中获取任务，任务不为空则一直循环
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())
// 如果线程池状态 >= STOP，或者如果线程池状态 >= STOP并且当前线程没有中断，终端当前线程
wt.interrupt();
try {
// 空方法，可以自己定义
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 执行run方法，其实就是Runable接口中的方法实现
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 {
// 任务完成后，任务置空，任务完成数累加1，释放锁
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 删除已完成的任务
processWorkerExit(w, completedAbruptly);
}
}
源码执行流程图如下：
微信图片_20190529094340.jpg
