1.前言
本章本来要讲解Netty的线程模型的，但是由于其是基于Java线程池设计而封装的，所以我们先详细学习一下Java中的线程池的设计。之前也说过Netty5被放弃的原因之一就是forkjoin结构比较复杂，forkjoin也是JDK提供的一个基本线程模型，这里就不进行介绍。本节涉及知识点很多，可能有误，请对照JDK源码进行学习。
本章涉及的概念有Callable，Future，ExecutorService等，所有的类都在java.util.concurrent包下。
2.相关概念
2.1 Callable
学过Java的人都知道，在Java中运行线程任务有两个方法，一个是继承Thread类，覆写run方法，另一个就是实现Runnable接口，实现run方法。但是这两种方法都有一个尴尬的地方，其没有返回值，而通常我们使用异步模型，需要知道异步的任务的执行结果，虽然方法有很多，但是最简单的还是重新制造一个带返回值的接口，这个接口就是Callable。此接口就一个call()方法，返回一个结果。此外Thread只接受Runnable类，所以该接口是专门给线程池模型使用的。
2.2 Future
Future顾名思义，这个类的实例是从未来而来。异步模型，在主线程中提交任务到线程池中运行，得到执行后的结果，问题是主线程如何获取这个结果呢？这个时候就要靠这个来自未来的实例了。提交任务给异步线程时，会立刻得到一个Future对象，这个对象可以获得任务执行完成后的结果。明白了这个设定，就能很好的理解接口方法了。
cancel()：取消掉这个任务，true时意味着运行中中断程序。实际上实现是办不到的，这个要自己实现的call方法做相应的取消实现，所以其正在的含义是改变Thread的状态成中断状态，具体中断处理逻辑要自己写在call或者run方法中。false则只是取消任务，不能对开始了的任务进行操作。
isCancelled()：这个任务是否被取消
isDone()：这个任务是否被完成
get()：获取任务的执行结果，没完成就阻塞获取线程，在设置的时间内没获取到抛出TimeoutException。
2.3 ExecutorService
Java设计了很多种类型的线程模型，开发者可能扩展更多，为了有个统一的标准，就有了这个接口。
shutdown()：关闭模型中所有的线程
isShutdown()：是否已经关闭
isTerminated()：所有的任务是否在shutdown之后终结了。
awaitTermination()：等待任务结束。
submit()：提交一个任务
invokeAll()：执行所有的任务
invokeAny()：执行所有的任务，其中一个完成就返回结果。
ExecutorService继承自Executor，其还有一个方法就是execute执行一个runnable任务。从这些接口方法，我们也可以清楚的认识到该类是对线程模型的一个基本使用的定义，提供了一个运用标准。
2.4 Executors
该类是一个具体的服务层的类了，给出所有JDK实现的线程模型的实例，方便开发者调用。该类的所有方法都是静态方法，静态工厂模式。
给出的方法有很多，但实际上只有三种类型的模型而已：1.线程池ThreadPoolExecutor；2.工作窃取算法实现的ForkJoinPool；3.周期任务线程池ScheduledThreadPoolExecutor；下面对这些方法进行简要说明。这里先简单介绍一下ThreadPoolExecutor的相关参数含义，才能更好的理解Executors提供的不同线程池的含义，具体参数如下：
corePoolSize：核心线程池的大小，可以理解为线程池中最小的线程数量。
maximumPoolSize：最大的线程数量
keepAliveTime：线程池线程数量超过core的设置后，线程最大的空闲时间，超过这个值且数量仍大于core会被销毁。
unit：keepAliveTime的单位
workQueue：任务的存储队列
threadFactory：创建线程时的工厂方法
handler：执行阻塞时处理类，当线程数达到最大，任务存储队列也满了，新提交的任务会触发该方法。
现在我们来看看几种和ThreadPoolExecutor相关的方法：
1.newFixedThreadPool(nThreads)：
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
从参数可以看出，FixedThreadPool是一个可重用固定线程数的线程池，由于固定了线程数量，所以keepAliveTime参数无实际作用。其任务在LinkedBlockingQueue中，这是一个无边界的队列，所以handler无作用。
2.newSingleThreadExecutor()：
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
可以看出，SingleThreadExecutor和FixedThreadPool参数基本一样，就是线程数量为1，keepAliveTime和handler参数不起作用。
3.newCachedThreadPool()：
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
CachedThreadPool是一个无限制的线程池，其线程空闲1分钟就会被回收，SynchronousQueue在之前的博客中也介绍过：这里。这是个无容量的阻塞队列，有点像接力棒。由于线程池大小没有限制，所以实际上handler对该类也没有作用。
4.newScheduledThreadPool(int corePoolSize)：
return new ScheduledThreadPoolExecutor(corePoolSize);
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
定时任务线程池是继承自ThreadPoolExecutor的，其能做到定时执行，归功于DelayedWorkQueue。这个队列功能和DelayQueue类似：这里，设计实现上与PriorityBlockingQueue队列相似：这里。这个线程池也是无边界的。
5.newSingleThreadScheduledExecutor()：
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
这个和newSingleThreadExecutor一样，就是单线程的周期任务。
6.newWorkStealingPool()：
return new ForkJoinPool
(Runtime.getRuntime().availableProcessors(),
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, true);
ForkJoin的核心思想就是：将一个大任务切割成若干个小任务同时进行，最后等所有任务完成，合并任务结果。工作窃取算法是其的一种优化，比如有个线程执行完了其任务，另一个线程还有N个任务，这样等待就比较耗时，所以空闲的线程会从忙碌的线程处理的任务链尾端拿任务进行执行。
其它的不是线程模型相关的内容，值得一提的就只有RunnableAdapter了。
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
实现简单易懂，不再描述。
3. 实现细节
上面讲了线程池的一些基本概念，这里将对其实现过程进行解析，理解更深刻。上面的概念有两个地方没有说明清楚，一个是Future为什么能达到这种效果。另一个就是callable接口为什么能返回结果，我们知道Thread只接收Runnable接口实例，该方法即没有入参，又没有返回值，这是怎么做到的？带着这两个疑问，我们对ThreadPoolExecutor进行详细的介绍。
3.1 FutureTask
这个类是解决上诉问题的关键。其实现了Future和Runnable两个接口，是线程池使用的具体任务对象，控制着任务的相关执行。
state：任务的状态。后面是其7种状态定义。
callable：需要执行的任务，如果是runnable会变成callable的适配器对象。
outcome：执行结果
runner：执行的线程
waiters：等待任务执行完成的线程
public boolean isCancelled() {
return state >= CANCELLED;
}
public boolean isDone() {
return state != NEW;
}
上面很清楚，状态就是由该类控制的。
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
这里就get方法的逻辑：1.判断线程是否中断，中断移除无效等待节点；2.如果完成了，返回；3.如果进行中，让出时间片；4.如果无等待者，将此线程设置成waiter；5.如果没添加到task的等待节点中，添加进去；6.如果有超时设置，超时了移除等待，否则等待。7.阻塞等待执行完成。
简要的说一下上述过程就是根据state判断是立刻返回还是阻塞等待，等待后将其设置成等待节点。最后我们需要关心的就是run方法了。
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
判断状态，执行callable的call方法，获取返回值。这里可能会有个疑问，runnable接口是怎么返回结果的呢？答案是runnable方法不能返回结果，但是可以通过RunnableAdapter传入一个默认的结果，在runnable任务结束时就能获取这个结果了。上面的问题2实际上要说明的就是没有其他方法对一个既没有入参，又没有出参的方法进行返回，最好的方法就是对其包装一层，通过包装的局部变量返回执行结果。FutureTask就是这么做的，最后看下任务完成后执行的方法。
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
这里就能够看出，在get操作任务未完成时，get线程阻塞，并且将阻塞线程添加到任务的waiter队列中。任务实际完成后，会释放锁，这就达到通知线程任务完成的效果了。
3.2 AbstractExecutorService
线程池的任务定义，通知任务完成的实现原理等了解完了，现在再来看下线程池是如何实现的吧。该类为抽象父类，没有做核心的操作，关键的execute方法没有实现，实现了其它通用逻辑的方法。简略看下就可以明白了：
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {
if (tasks == null)
throw new NullPointerException();
int ntasks = tasks.size();
if (ntasks == 0)
throw new IllegalArgumentException();
ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
ExecutorCompletionService<T> ecs =
new ExecutorCompletionService<T>(this);
// For efficiency, especially in executors with limited
// parallelism, check to see if previously submitted tasks are
// done before submitting more of them. This interleaving
// plus the exception mechanics account for messiness of main
// loop.
try {
// Record exceptions so that if we fail to obtain any
// result, we can throw the last exception we got.
ExecutionException ee = null;
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Iterator<? extends Callable<T>> it = tasks.iterator();
// Start one task for sure; the rest incrementally
futures.add(ecs.submit(it.next()));
--ntasks;
int active = 1;
for (;;) {
Future<T> f = ecs.poll();
if (f == null) {
if (ntasks > 0) {
--ntasks;
futures.add(ecs.submit(it.next()));
++active;
}
else if (active == 0)
break;
else if (timed) {
f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
if (f == null)
throw new TimeoutException();
nanos = deadline - System.nanoTime();
}
else
f = ecs.take();
}
if (f != null) {
--active;
try {
return f.get();
} catch (ExecutionException eex) {
ee = eex;
} catch (RuntimeException rex) {
ee = new ExecutionException(rex);
}
}
}
if (ee == null)
ee = new ExecutionException();
throw ee;
} finally {
for (int i = 0, size = futures.size(); i < size; i++)
futures.get(i).cancel(true);
}
}
这里就能看出提交的任务都被包装成FutureTask任务了。invokeAll方法就是包装所有的任务，遍历调用execute方法。invokeAny最终也是相似的操作。
3.3 ThreadPoolExecutor
抽象父类的作用不多，这里我们看ThreadPoolExecutor是如何管理线程池，进行调度的吧。ThreadPoolExecutor的策略如下：
1、当有一个新任务提交时，只要当前线程数少于corePoolSize，哪怕有空闲的线程，其也会创建一个新的线程给新任务使用。
2、corePoolSize<X<maxSize的时候，只有队列满了才会创建线程。
3、只有有任务到来时才会创建线程，哪怕是corePoolSize也不是立刻初始化。
4、创建线程失败，执行器也会继续下去，但是任务不会被执行
5、根据上面的说明，无界的队列会导致线程数达到corePoolSize就不再增长，因为队列永远不会满
6、有界队列可以防止资源耗尽，但是更难正确和控制。
workQueue：任务队列
mainLock：主要的锁
worker：工作者，运行中线程
termination：终止信号
其他的就不一一介绍了，都是之前提到过的参数。这里直接看最重要的execute方法实现。
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();
}
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);
}
源码注释说明了这段操作，一共有3个步骤：
1.如果运行中的线程数小于corePoolSize的数量，创建一个新的线程，该任务作为其第一个任务。addWorker会检查池的状态和工作线程数量，不满足条件就会返回false。
2.如果一个任务成功的放入队列，再次检查状态，如果池停止了，就会拒绝该任务，且移除该任务，否则触发方法，请求分配worker。
3.如果不能放入队列，尝试创建一个线程，如果失败，就拒绝该任务。
可以说思路其实是很清晰的，但是实际上操作却不容易，具体细节就不说明了。接下来看下Worker是如何工作的：
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
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);
}
}
worker也是一个runnable方法，不同的是其继承了AbstractQueuedSynchronizer，该类是Java并发包的一个重点。创建worker对象的时候就初始化了线程设置，只等addWorker()方法调用start()。运行过程判断有没有任务，有任务锁住线程，然后运行任务。这里都有不同的扩展点给开发者进行业务扩展。完成任务后，在运行中的线程集合中，移除它。这里可能会疑惑，运行完后线程被移除了，怎么继续执行的任务。
首先，runWorker中task为null的时候会getTask，从任务队列中获取任务。
其次，execute方法判断了当前池的状态决定是否添加worker。
然后，最重要的是runWorker方法要结束必须是没有任务，那里有个while循环。
最后，该worker被移除，检测corePoolSize和线程池当前状态决定是否添加线程。
这里又可能会有疑惑了，不是说好的空闲超时才移除吗？这段逻辑其实就在getTask中，如果getTask在指定时间内都没有获取任务，不就意味着线程空闲了这么久吗，所以执行到最后直接移除就可以了。最后我们看下Java的线程池是如何关闭的，这对后面理解Netty的优雅停机有帮助。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
先解决权限问题，再设置状态为shutdown，这样保证新任务不会再提交进线程池。接着对所有运行中的worker发起中断信号，最后尝试停机，唤醒awaitTermination的线程。所以说所谓的shutdown，并不保证任务执行完毕，只是阻止新任务进来，还需要任务自身配合中断信号。这个是必然的，因为没人知道你的任务干了什么，万一是个死循环，不直接销毁线程，如何能停止它？
最后我们聊聊AbstractQueuedSynchronizer，每一个worker都是该类实例，这个类究竟达成了一个怎样的目的？详细介绍看这篇文章：这里。下面是Worker中实现的相关代码：
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
根据文章中所介绍的：
- tryAcquire：独占式获取同步状态。获取时设置状态为1。
- isHeldExclusively： AQS是否被当前线程独占。不为0独占。
- tryRelease：独占式释放同步状态。设置状态为0，释放锁。
所以worker类实际上是一个排他锁，只允许一个线程操作。
4 后记
本章主要介绍了Java线程池的执行过程，Future的实现方式，整个过程可用下图表示：
再提一下FutureTask，其call执行完就会解锁，get就能知道是否执行完毕了。大体的过程就是上图，实现细节就不在此讨论。