ThreadPoolExecutor是怎样执行任务的
先扯点别的：最近被公司外派到了嘉定区，新租了一个房子，马马虎虎，每天走路上班大约30分钟。早上8点半上班，下午5点半下班，和以前的作息规律有点不一样，逐渐适应，调整，然后还得把跑步这项运动坚持下来。
这篇文章（一天写不完）探索一下Java中的ThreadPoolExecutor的使用，感觉还是挺重要的。如果出去面试能把这个讲清楚，估计能唬住不少面试官，哈哈。
先给出一个简单的使用示例，然后再探索源码。
ThreadPoolExecutor的继承结构
使用示例
public class ThreadPoolTest {
public static void main(String[] args) {
System.out.println(Thread.currentThread().getName());
//创建一个ThreadPoolExecutor
ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
for (int i = 0; i < 15; i++) {
MyTask myTask = new MyTask(i);
//执行任务
executor.execute(myTask);
}
//所有的任务提交以后终止ThreadPoolExecutor
executor.shutdown();
}
}
//要被执行的任务类
class MyTask implements Runnable {
private int taskNum;
public MyTask(int taskNum) {
this.taskNum = taskNum;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " 正在执行task" + taskNum);
try {
Thread.sleep(4000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " task" + taskNum + "执行完毕");
}
}ThreadPoolExecutor的使用简单说就是分三步走。
1. 新建ThreadPoolExecutor实例。
2. ThreadPoolExecutor 执行任务。
3. ThreadPoolExecutor在适当的时候终止。
源码探索
新建ThreadPoolExecutor实例
一般可以使用Executors 类中的静态方法来创建ThreadPoolExecutor实例。
//创建一个可固定数量线程的线程池和一个无限队列
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//创建只有一个线程的线程池和一个无限队列
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//创建可以有无限多个线程的线程池和一个同步队列（如果想向队列中加入元素，队列必须是空的，如果不为空，那么添加操作会阻塞。如果想从队列中取数据，如果队列为空，则阻塞）
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
上面的三个方法各有一个重载方法，可以用于指定ThreadFactory 。上面的方法内部还是调用了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;
}构造函数中参数的含义
1 corePoolSize：核心线程池大小,即使核心线程是空闲状态也不会被销毁。除非设置allowCoreThreadTimeOut为true。创建一个线程池以后并不会创建任何新的线程，
而是等待有任务到来才去创建线程执行任务。可以调用prestartAllCoreThreads()方法或者prestartCoreThread()方法预先创建corePoolSize个或者1个线程，等待任务的到来。
2 maximumPoolSize：线程池最大线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多少个线程。
3 keepAliveTime：表示线程空闲多长时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize。即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0。
4 unit：keepAliveTime的单位。
5 workQueue：阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择ArrayBlockingQueue LinkedBlockingQueue SynchronousQueue 。
6 threadFactory：线程工厂，主要用来创建线程。
7 handler：表示当拒绝处理任务时的策略，有以下四种取值
- ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
- ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
- ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）。
- ThreadPoolExecutor.CallerRunsPolicy：由调用
ThreadPoolExecutor.execute()方法所在的线程执行该任务 。
行，ThreadPoolExecutor实例创建就先到这。下面看ThreadPoolExecutor 是怎么执行任务的。
ThreadPoolExecutor 有两个方法可以用来执行任务。
1 Future<?> submit(Runnable task)方法：这个方法需要配合FutureTask使用，可以获取任务执行后返回的结果。这个方法是在ExecutorService 接口中定义的，在AbstractExecutorService 类中就有了具体的实现。
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
//内部还是调用了execute来执行任务。
execute(ftask);
return ftask;
} ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
FutureTask<Integer> futureTask = new FutureTask<>(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
return 100;
}
});
executor.submit(futureTask);
Integer result = futureTask.get();
System.out.println(result);2 void execute(Runnable command)方法。这个方法是在Executor接口中定义的，在ThreadPoolExecutor 类中有具体的实现。在介绍这个方法之前，先介绍一下ThreadPoolExecutor 中比较重要的变量。
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));//标志当前线程池的状态
private static final int COUNT_BITS = Integer.SIZE - 3;//等于29
//用二进制表示CAPACITY 0001 1111 1111 1111 1111 1111 1111 1111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//线程池的状态
//用二进制表示RUNNING 1010 0000 0000 0000 0000 0000 0000 0000
private static final int RUNNING = -1 << COUNT_BITS;
//用二进制表示SHUTDOWN 0000 0000 0000 0000 0000 0000 0000 0000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//用二进制表示STOP 0010 0000 0000 0000 0000 0000 0000 0000
private static final int STOP = 1 << COUNT_BITS;
//用二进制表示TIDYING 0100 0000 0000 0000 0000 0000 0000 0000
private static final int TIDYING = 2 << COUNT_BITS;
//用二进制表示TERMINATED 0110 0000 0000 0000 0000 0000 0000 0000
private static final int TERMINATED = 3 << COUNT_BITS;
private final BlockingQueue<Runnable> workQueue;//任务缓存队列，用来存放等待执行的任务
private final ReentrantLock mainLock = new ReentrantLock();//操作许多变量都需要这个锁
private final HashSet<Worker> workers = new HashSet<Worker>();//存放工作集，需要获取mainLock才可以操作这个变量
private volatile boolean allowCoreThreadTimeOut;//是否允许为核心线程设置存活时间
private int largestPoolSize;//用来记录线程池中曾经出现过的最大线程数
private long completedTaskCount;//用来记录已经执行完毕的任务个数几个用来获取线程池状态和线程池中线程数量的方法
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }初始化的时候线程池的状态是 RUNNING
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));我们使用newSingleThreadExecutor 这种类型的线程池来分析上面的方法。这种线程池内部只有一个线程和一个无LinkedBlockingQueue 。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
} ExecutorService executor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
MyTask myTask = new MyTask(i);
executor.execute(myTask);
}
executor.shutdown();MyTask 类已经在上面定义过了，就不在贴代码了。
pool-1-thread-1 正在执行task0
pool-1-thread-1 task0执行完毕
pool-1-thread-1 正在执行task1
pool-1-thread-1 task1执行完毕
pool-1-thread-1 正在执行task2
pool-1-thread-1 task2执行完毕
pool-1-thread-1 正在执行task3
pool-1-thread-1 task3执行完毕
pool-1-thread-1 正在执行task4
pool-1-thread-1 task4执行完毕
pool-1-thread-1 正在执行task5
pool-1-thread-1 task5执行完毕
pool-1-thread-1 正在执行task6
pool-1-thread-1 task6执行完毕
pool-1-thread-1 正在执行task7
pool-1-thread-1 task7执行完毕
pool-1-thread-1 正在执行task8
pool-1-thread-1 task8执行完毕
pool-1-thread-1 正在执行task9
pool-1-thread-1 task9执行完毕
Process finished with exit code 0
可以看到执行的结果是按照任务的提交顺序执行的。
接下来，我们分析一下 execute(Runnable command) 方法
public void execute(Runnable command) {
//任务不能为空
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//如果当前工作线程数量，小于核心线程的数量，就添加一个新的工作线程，并把任务作为新添加线
//程的第一个任务,如果添加成功就返回
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
//添加失败，再次获取线程池的状态
c = ctl.get();
}
//如果线程池状态是RUNNING 并且任务成功添加到队列中
if (isRunning(c) && workQueue.offer(command)) {
//再次检查
int recheck = ctl.get();
//如果线程池状态不是RUNNING（就是说线程池关闭了），并且能成功的把任务从队列中移除
if (! isRunning(recheck) && remove(command))
//拒绝任务
reject(command);
//如果当前没有工作线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果不能把任务加入到队列，那就添加一个工作线程，如果添加失败（线程池已经shut down 或者已经饱和了），就拒绝任务。
else if (!addWorker(command, false))
reject(command);
}上面代码执行的步骤使用文字来叙述一下。
1 如果当前工作线程数量，小于核心线程的数量，就添加一个新的工作线程，并把任务作为新添加程的第一个任务,如果添加成功就返回。
2 如果当前线程池状态是RUNNING，任务可以成功加入到队列中，那么我们仍然需要检查我们是否应该添加一个线程（因为现存的工作线程可能自从上次检查后死亡）或者进入该方法以后，线程池关闭了。这时候分两种情况
- 第一种情况：如果线程池关闭了，并且能成功的把任务从队列中移除，那么就拒绝这个任务。
- 如果第一种情况不满足，那么就判断，如果线程池中没有工作线程，就添加一个线程。
3 添加一个工作线程，如果添加失败（线程池已经shut down 或者已经饱和了），就拒绝任务。
上面的三条是对这个方法的大致流程的一个顺序，接下来，看一下方法中调用的addWorker方法。
private boolean addWorker(Runnable firstTask, boolean core) {
//break标签
retry:
//外层循环
for (;;) {
int c = ctl.get();
//当前线程的状态
int rs = runStateOf(c);
// 如果线程池已经已经STOP,或者TIDYING，或者TERMINATED，并且，队列并不为空的
//情况下，就直接返回false。意思就是线程池说：老子要下班了，已经在队列里等待执行的
//任务都不执行了，你这个心任务还想被执行，拉倒吧你。
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//内层循环
for (;;) {
int wc = workerCountOf(c);
//当线程池中的线程数量达到上限以后，返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//如果线程数量能够增加，则break到retry标签，（跳过内外层两层循环），继续往下执行。
if (compareAndIncrementWorkerCount(c))
break retry;
//重新获取线程池的状态
c = ctl.get(); // Re-read ctl
//如果线程池的状态发生了改变，就continue到retry标签，重新执行循环。
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
//如果CAS操作失败，是因为工作线程数量的变化，那么继续执行内层循环
}
}
//标记新的工作线程是否添加成功
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;
mainLock.lock();
try {
//获取锁以后再次检查
int rs = runStateOf(ctl.get());
//如果线程池状态是RUNNING 或 线程池状态是SHUTDOWN 并且fistTask==null
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 表示线程已经被启动过了，则抛出异常
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 {
//最后判断启动任务如果失败，就把任务从工作集中移除，减小最大线程数largestPoolSize，并试着终止线程池
if (! workerStarted)
addWorkerFailed(w);
}
//返回任务是否成功启动
return workerStarted;
}上面的注释已经很清楚了，这里再梳理一下流程。
1 如果线程池已经已经STOP,或者TIDYING，或者TERMINATED，并且，队列并不为空的情况下，就直接返回false。意思就是线程池说：老子要下班了，已经在队列里等待执行的任务都不执行了，你这个心任务还想被执行，拉倒吧你。
2 当线程池中的线程数量达到上限以后，返回false，否则循环，直到能够增加线程数量。
3 如果能够增加线程池数量，就新建一个Worker，加入工作集，改变最大线程数量largestPoolSize 。添加成功就启动线程，如果启动线程失败，就把Worker 从工作集中移除，减小largestPoolSize，并试着终止线程池。最后返回任务是否成功启动。
我们看一下Worker这个类
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
//执行当前任务的线程
final Thread thread;
//要执行的任务，可能为null
Runnable firstTask;
//每个线程的任务计数器，用来标记当前线程执行了多少个任务
/** Per-thread task counter */
volatile long completedTasks;
//构造函数
Worker(Runnable firstTask) {
...
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
//run方法调用了ThreadPoolExecutor的runWorker方法
public void run() {
runWorker(this);
}
...
//省略其他方法
}看一下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不为null或者从队列中取出的任务不为null
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);
}
}protected void beforeExecute(Thread t, Runnable r) { }我们可以重写这个方法，在执行每个任务之前做一些操作。
protected void afterExecute(Runnable r, Throwable t) { }我们可以重写这个方法，在执行每个任务之后做一些操作。
先分析到这里，以后有新的理解和体会，在继续写。
现在总结一下 线程池执行任务的流程：
1.如果当前线程池中的线程数目小于corePoolSize，则每来一个任务，就会创建一个线程去执行这个任务；
2.如果当前线程池中的线程数目>=corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列当中，若添加成功，则该任务会等待空闲线程将其取出去执行；若添加失败（一般来说是任务缓存队列已满），则会尝试创建新的线程去执行这个任务；
3.如果当前线程池中的线程数目达到maximumPoolSize，则会采取任务拒绝策略进行处理；
结尾：若有不正之处请多多谅解，并欢迎批评指正。
上一篇： mybatis中使用in查询时的注意事项
下一篇： 梦幻式下拉菜单