一、回顾：
线程运行的两种方式：
方法一：
- 创建自定义类extends Thread （或者使用匿名类进行定义）
- 重写run()方法
- 调用该类start方法启动
new Thread(){
public void run(){
System.out.println(getName()+":正在执行！");
}
}.start();
方法二：
- 创建自定义类implements Runnable接口
- 重写Runnable接口中的run()方法
- 在主线程中创建一个Thread对象并且将自定义类对象传入
- 使用Thread对象开启start()方法
new Thread(new Runnable(){
public void run(){
System.out.println("正在执行！");
}
}).start();
以上两种方法都是将每个任务放在各自的线程中去运行，然而这会存在一些问题，尤其是当需要创建大量的线程时：
- 线程的创建和销毁开销非常的高
- 资源消耗：如果可运行的线程数量多于可用处理器的数量，大量的空闲线程会占用许多内存，且大量线程在争夺cpu资源的时候会产生其他的性能开销
为了避免无限制的创建线程，Java提供了第三种多线程运行的方法：线程池
二、线程池的相关类和接口
Executor接口
在Java类库中，任务执行的主要抽象不是Thread，而是Executor，Executor是一个简单接口
public interface Execurot{
void execute(Runnable command);
}
Executor框架提供了一种标准的方法将任务的提交过程与执行过程解耦开来，并用Runnable来表示任务。我们可以通过改变Executor的实现或配置来改变线程的执行方式。
ExecutorService接口
Executor的实现通常会创建线程来执行任务，但JVM只有在所有（非守护）线程全部终止后才会退出，因此，如果无法正确的关闭Executor，那么JVM将无法结束。
为了解决执行服务的生命周期问题，ExecutorService扩展了Executor接口，添加了一些用于生命周期管理的方法
public interface ExecutorService extends Executor{
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout,TimeUnit unit) throws INterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
// ...其他用于任务提交的便利方法
}
ExecutorService的生命周期有三种状态：运行、关闭和已终止，shutdown平缓关闭，不再接受新的任务，同时等待已经提交的任务执行完成——包括还未开始执行的任务。shutdownNow方法粗暴的关闭，尝试取消所有正在运行的任务。在ExecutorService关闭后提交的任务将由Rejected Execution Handler来处理，它会抛弃任务，或者使得execute方法抛出一个未检查的Rejected ExecutionException。
可以使用awaitTermination等待ExecutorService到达终止状态，或者调用isTerminated来查询时候已经终止。
除此之外，ExecutorService还提供了submit接口，返回一个Future对象，可以通过对Future对象调用get方法得到结果。
Callable与Future
Runnable接口是一种有很大局限的抽象，它不能返回一个值或抛出一个受检查的异常。Callable是一种更好的抽象，它可以返回一个值，也可以抛出一个异常，执行Callable任务后，可以获取一个Future对象。Future表示一个任务的生命周期，并提供了相应的方法来判断是否已经完成或取消，以及获取任务的结果和取消任务等。
在该对象上调用get就可以获取Callable任务返回的Object了。再结合线程池接口ExecutorService提供的submit方法就可以实现传说中有返回结果的多线程。
public interface Callable<V>{
V call() throws Exception;
}
public interface Future<V>{
boolean cancel(boolean myInterruptIfRunning);
boolean isCanceller();
boolean isDone();
V get() throws InterruptedException, ExecutionException,CancellationException;
V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, CancellationException, TimeoutException;
}
get方法的行为取决于任务的状态（尚未开始，正在运行，已完成）如果任务已经完成，那么get会立即返回或者抛出一个Exception，否则将阻塞并直到任务完成。可以通过许多方法创建一个Future来描述任务，ExecutorService中的所有submit方法都将返回一个Future
三、ThreadPoolExecutor线程池
线程池实现了ExecutorService接口，也即拥有ExecutorService的功能。有两种方法创建一个线程池，使用Executors这一线程池工厂类创建默认的几种线程池，或者自定义线程池。自定义一个线程池需要定义7个参数，默认的线程池无非是在这8个参数中赋上了一定的值。
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
* parameters and default thread factory.
*
* @param corePoolSize the number of threads to keep in the pool, even
* if they are idle, unless {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the
* pool
* @param keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are
* executed. This queue will hold only the {@code Runnable}
* tasks submitted by the {@code execute} method.
* @param handler the handler to use when execution is blocked
* because the thread bounds and queue capacities are reached
*/
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
线程池中维护着两个队列，线程队列和任务队列，运行顺序是当任务来临时，先调用核心线程去执行，如果核心线程数不够，则将任务放入任务队列中，如果任务队列已满，且线程数小于最大线程池大小，则开启新的非核心线程，这类线程在空闲时间达到keepAliveTime之后会自动释放掉，如果任务队列已满且线程数也达到了最大线程池大小，则执行拒绝策略
设置线程池的大小
如果线程池过大，大量的线程在相对很少的CPU和内存资源上发生竞争，导致更高的内存使用量以及CPU切换开销；如果线程池过小，将会导致许多空闲的处理器无法执行工作，从而降低吞吐量。如何合理的设置线程数，这里有一个公式：
举一个例子可以更好的理解这个公式，有一个八核的cpu，处理http一个请求需要消耗cpu时间20ms，进行io相关操作需要消耗180ms，请问需要开多少线程。可以这样想，如果想让cpu使用率最大化，那么应该在处理io请求，cpu处于空闲状态时再开一个线程处理其他的cpu请求，那么在180ms的时间里可以处理多少个cpu请求呢？答案是9个，加上最初的一个线程，在处理一个http请求的200ms的周期中，10个线程就可以让cpu的一个核心全速运行，那么8个核心一共需要开启80个线程。
如果我们不想让cpu的占用率100%，那么可以这么考虑，我们不想在180ms的cpu空闲时间内将cpu全部占用起来，也即占用率低于100%，那么可以理解为在200ms的周期中，让cpu工作占用率%的时间，比如想让cpu的一个核心的占有率为50%，那么cpu一共工作的时间为100ms，100ms/20ms=5个线程，所以8个核心就需要40个线程。上面的这个公式就是这个意思。
阻塞队列：
BlockingQueue接口继承了Queue接口，它提供了可阻塞的put和take方法，以及支持定时的offer和poll方法。BlockingQueue天然的实现了生产者-消费者模式，如果队列已满，put操作将会被阻塞，如果队列已空take操作将被阻塞，其本质上是维护了两个信号量notEmpty和notFull，以ArrayBlockingQueue为例
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}
基本的任务排队方式有三种：无界队列、有界队列和同步移交，无界队列以LinkedBlockingQueue为代表的，有界队列以ArrayBlockingQueue为代表的，同步移交则以SynchronousQueue为代表，SynchronousQueue的长度为0，要将一个元素放入SynchronousQueue中，必须有另一个线程正在等待接受这个元素。如果想进一步控制任务执行顺序，还可以使用PriorityBlockingQueue，这个队列根据优先级来安排任务
饱和策略：
如果有界队列被填满，且线程数量达到上限，饱和策略开始生效，JDK提供了几种RejectedExecutionHandler的实现，每种实现都有不同的饱和策略：AbortPolicy、CallerRunsPolicy、DiscardPolicy和DiscardOldestPolicy，这几个类都是ThreadPoolExecutor类中的子类
AbortPolicy是默认饱和策略，当饱和时会抛出非必检的RejectedExecutionException，调用者可以捕获这个异常，然后编写自己的处理代码。
DiscardPolicy会悄悄抛弃这个任务；DiscardOldestPolicy会抛弃下一个将被执行的任务；Caller-Runs策略不会抛弃任务，也不会抛出异常，而是将某些方法回退到调用者。
线程工厂：
可以通过实现ThreadFactory接口，重写newThread方法，定制一个线程工厂方法，每当线程池需要创建一个线程时会调用该方法。下面这个例子通过自定义线程工厂以及自定义线程，为线程指定了名字、维护了一些统计信息（创建了多少线程，多少线程存活）
public class MyThreadFactory implements ThreadFactory{
private final String poolName;
public MyThreadFactory(Stirng poolName){
this.poolName=poolName;
}
public Thread newThread(Runnable runnable){
return new MyAppThread(runnable,poolName);
}
}
public class MyAppThread extends Thread{
public static final String DEFALT_NAME="MyAppThread";
private static final AtomicInteger created=new AtomicInteger();
private static final AtomicInteger alive=new AtomicInteger();
public MyAppThread(Runnable r){
this(r,DEFALUT_NAME);
}
public MyAppThread(Runnable runnable,String name){
super(runnable,name+"-"+created.incrementAndGet());
setUncaughtExceptionHandler(
new Thread.UncaughtExceptionHandler(){
public void uncaughtException(Thread t,Throwalbe e){
System.out.println("未捕获的异常");
}
}
);
}
public void run(){
try{
alive.incrementAndGet();
super.run();
}finally{
alive.decrementAndGet();
}
}
public static int getThreadsCreated(){return created.get()};
public static int getThreadAlive(){return alive.get()};
}
创建默认的线程池：
可以使用Executors中的工厂方法，创建几种默认的线程池，通过查看源码，可以很轻松的读懂每一种默认线程池的特性
newFixedThreadPool：固定长度的线程池，无界队列
newCachedThreadPool：线程池规模不存在限制，任务队列使用SynchronousQueue进行管理
newSingleThreadPool：见闻知意，线程池是固定的只有1个线程，任务队列是无界的
newScheduledThreadPool: 延迟线程池，可以让使任务在延迟指定时间后再执行，起继承关系如下图所示
它使用ScheduledExecutorService接口提供的schedule方法提交任务。
附录 ThreadPoolExecutor源码分析
1、常用变量的解释
// 1. `ctl`，可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// 2. `COUNT_BITS`，`Integer.SIZE`为32，所以`COUNT_BITS`为29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 3. `CAPACITY`，线程池允许的最大线程数。1左移29位，然后减1，即为 2^29 - 1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
// 4. 线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
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;
// Packing and unpacking ctl
// 5. `runStateOf()`，获取线程池状态，通过按位与操作，低29位将全部变成0
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 6. `workerCountOf()`，获取线程池worker数量，通过按位与操作，高3位将全部变成0
private static int workerCountOf(int c) { return c & CAPACITY; }
// 7. `ctlOf()`，根据线程池状态和线程池worker数量，生成ctl值
private static int ctlOf(int rs, int wc) { return rs | wc; }
/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/
// 8. `runStateLessThan()`，线程池状态小于xx
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// 9. `runStateAtLeast()`，线程池状态大于等于xx
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
2、构造方法
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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
// 根据传入参数`unit`和`keepAliveTime`，将存活时间转换为纳秒存到变量`keepAliveTime `中
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
3、提交执行task的过程
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);
}
4、addworker源码解析
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;
}
5、线程池worker任务单元
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);
}
// 省略代码...
}
6、核心线程执行逻辑-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);
}
}