线程池深入解析笔记

https://www.jianshu.com/p/117571856b28

https://juejin.im/entry/58fada5d570c350058d3aaad

概述

这篇笔记也是整理于寒假实习期间，由于美颜相机demo中的缩略图渲染功能涉及到多线程，使用了线程池减少资源开销，所以整理一下线程池笔记，以便以后查询（待完善）。

使用线程池的主要目的在于：

1. 降低资源消耗

2. 提高响应速度

3. 提高线程的可管理性

几个参数

线程池的构造器中主要有以下几个参数

1. corePoolSize：核心线程的数量。核心线程是在执行完任务之后也不会被销毁的线程。

2. maximumPoolSize：线程池中的最大线程数。表示线程池中最多能同时存在多少个线程。

3. keepAliveTime：表示线程没有任务执行时最多能存活多久。默认情况下只在线程数大于corePoolSize时起作用。但是通过调用allowCoreThreadTimeOut(boolean)方法可以在线程池中数量不超过corePoolSize时也会起作用，直到线程池中的线程数为0。

4. unit：keepAliveTime的时间单位，具体取值看源码。

5. workQueue：一个阻塞队列，用来存储等待执行的任务。一般情况下有以下几种队列：

* ArrayBlockingQueue：基于数组的有界阻塞队列，按照FIFO原则对元素排序。

* LinkedBlockingQueue：基于链表的无界阻塞队列，按照FIFO排序。Executors.newFixedThreadPool()使用了这个队列。

* SynchronizedQueue：一个不存储元素的阻塞队列，每个插入操作都必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态。Executors.newCachedThreadPool使用了这个队列。

* PriorityBlockingQueue：一个具有优先级的无限阻塞队列。

6. threadFactory：线程工厂，用于创建线程。

7. handler：饱和策略，即当队列和线程池都满了，应该采取什么策略来处理新提交的任务。通常有以下四种策略：

* ThreadPoolExecutor.AbortPolicy：丢弃任务并抛出RejectedExecutionException。

* ThreadPoolExecutor.DiscardPolicy：丢弃任务，但是不抛异常。

* ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务。

* ThreadPoolExecutor.CallerRunsPolicy：由调用线程执行任务。

线程池的原理

线程的状态

public enum State {
/**
* Thread state for a thread which has not yet started.
*/
NEW,

/**
* Thread state for a runnable thread.  A thread in the runnable
* state is executing in the Java virtual machine but it may
* be waiting for other resources from the operating system
* such as processor.
*/
RUNNABLE,

/**
* Thread state for a thread blocked waiting for a monitor lock.
* A thread in the blocked state is waiting for a monitor lock
* to enter a synchronized block/method or
* reenter a synchronized block/method after calling
* {@link Object#wait() Object.wait}.
*/
BLOCKED,

/**
* Thread state for a waiting thread.
* A thread is in the waiting state due to calling one of the
* following methods:
* <ul>
*   <li>{@link Object#wait() Object.wait} with no timeout</li>
*   <li>{@link #join() Thread.join} with no timeout</li>
*   <li>{@link LockSupport#park() LockSupport.park}</li>
* </ul>
*
* <p>A thread in the waiting state is waiting for another thread to
* perform a particular action.
*
* For example, a thread that has called <tt>Object.wait()</tt>
* on an object is waiting for another thread to call
* <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
* that object. A thread that has called <tt>Thread.join()</tt>
* is waiting for a specified thread to terminate.
*/
WAITING,

/**
* Thread state for a waiting thread with a specified waiting time.
* A thread is in the timed waiting state due to calling one of
* the following methods with a specified positive waiting time:
* <ul>
*   <li>{@link #sleep Thread.sleep}</li>
*   <li>{@link Object#wait(long) Object.wait} with timeout</li>
*   <li>{@link #join(long) Thread.join} with timeout</li>
*   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
*   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
* </ul>
*/
TIMED_WAITING,

/**
* Thread state for a terminated thread.
* The thread has completed execution.
*/
TERMINATED;
}

由Thread内部定义的枚举类型State可以看出线程共有六种状态：

1. NEW：新建线程，但是还没调用start()方法。处于NEW状态的线程有自己的内存空间。

2. RUNNABLE：可运行状态，线程对象创建且调用了start()方法后进入该状态，该状态的线程位于可运行线程池中，等待被线程调度器选中，获取CPU的使用权。

3. BLOCKED：等待获取锁时进入的状态，线程被挂起，通常是因为它在等待一个锁。当某个synchronized正好有线程在使用时，一个线程尝试进入这个临界区，就会被阻塞。当它抢到锁之后，才会从BLOCKED状态转换为RUNNABLE状态。

4. WAITING：当调用wait()、join()、park()方法时，进入WAITING状态。前提是这个线程已经拥有锁。

WAITING状态与BLOCKED状态的区别在于：①、BLOCKED是虚拟机认为程序还不能进入某个临界区。 ②、WAITING状态的先决条件是当前线程已经在临界区中，但是它自己通过判定业务上的参数，发现还有一些其他配合的资源没有准备充分而选择等待再做其他事情。

可以通过notify/notifyAll动作，从等待池中唤醒线程重新恢复到RUNNABLE状态。

5. TIMED_WAITING：通过wait(t)、sleep(t)、join(t)等方法进入此状态。当时间达到时触发线程回到工作状态RUNNABLE。interrupt只对WAITING或者TIMED_WAITING状态的线程起作用。

6. TERMINATED：线程结束后就是这种状态。也就是run方法执行完毕。

线程池的状态

// runState is stored in the high-order bits
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中定义的常量可以知道线程池由五种状态：

1. RUNNING：能接受新提交的任务，并且也能处理阻塞队列的任务。

2. SHUTDOWN：关闭状态，不再接受新的任务，但可以继续处理阻塞队列中一保存的任务。调用shutdown()方法会进入此状态。

3. STOP：不能接受新任务，也不能处理队列中的任务。调用shutdownNow()方法会进入此状态。

4. TIDYING：如果所有的任务都已终止，workerCount（有效线程数）为0，线程池会进入这个状态。只有会调用terminated()方法进入TERMINATED状态。

5. TERMINATED：终止状态。

状态转换图如下：

执行过程

我们通过submit(Runnable)方法来提交任务时，执行代码如下：

/**
* @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 ftas
e730
k;
}

它会把Runnable对象封装成一个RunnableFuture对象。然后调用execute方法。因此真正的执行逻辑在execute方法里：

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.
*/
//通过ctl.get()方法获取int值，记录了线程池当前的runState与workerCount。
int c = ctl.get();
//然后通过workerCountOf方法取出低29位的值，表示当前活动的线程数。
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
//如果添加失败，重新获取ctl值。
c = ctl.get();
}
//如果当前线程池是Running状态，且成功将任务添加进任务队列中
if (isRunning(c) && workQueue.offer(command)) {
//再次获取ctl值，如果不是运行状态，把刚添加进去的任务移除。
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
//调用reject方法，根据设定的拒绝策略执行不同的操作。
reject(command);
//如果当前有效线程数是0，则执行addWorker操作。
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//执行到这里，有两种情况：①、线程池已经不是RUNNING状态。②、线程池是RUNNING状态，但workerCount >= corePoolSize且workQueue已满。则再次执行addWorker方法。
else if (!addWorker(command, false))
//如果失败则拒绝执行任务。
reject(command);
}

相关解析看代码中的注释。

关于addWorker方法，有两个参数 Runnable firstTask 与 boolean core。第一个参数为null，表示在线程池中创建一个线程但不会启动。第二个参数为true，表示将线程池的线程数量上限设置为corePoolSize，为false表示将上限设置为maximumPoolSize。

线程池的调度方式

怎样开始执行

线程池通过addWorker方法成功创建一个新线程之后，会调用新线程的start方法。

private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);

// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;

for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
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;
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)) {
if (t.isAlive()) // precheck that t is startable
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 {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

t.start()句代码中，t实际是Worker中的Thread类型常量：

/** Thread this worker is running in.  Null if factory fails. */
final Thread thread;

thread是在Worker的构造方法中被初始化的：

/**
* 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;
this.thread = getThreadFactory().newThread(this);
}

在Worker的构造方法中，可以看到通过ThreadFactory创建了一个新的线程，同时传入了this，即是当前的worker对象。Woker类实现了Runnable接口，因此可以作为参数传入。这样，当在addWorker()方法中调用t.start()方法时，实际调用的就是Worker对象中的run()方法。那么接着看Worker中的run()方法：

/** Delegates main run loop to outer runWorker  */
public void run() {
runWorker(this);
}

里面只有一句，调用runWorker(this)。进入runWorker(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);
}
}

基本原理就是通过循环，不断地从任务队列中获取任务，并执行。在上面的源码中可以看到，beforeExecute()/afterExecute()方法也是在这里面被调用的。

怎样结束执行

AQS

戏（细）说Executor框架线程池任务执行全过程（下）

http://www.infoq.com/cn/articles/executor-framework-thread-pool-task-execution-part-02

线程池中的异常捕获

合理配置线程池（待整理）：

http://gityuan.com/2016/01/16/thread-pool/

4.1 合理地配置线程池

需要针对具体情况而具体处理，不同的任务类别应采用不同规模的线程池，任务类别可划分为CPU密集型任务、IO密集型任务和混合型任务。

对于CPU密集型任务：线程池中线程个数应尽量少，不应大于CPU核心数；

对于IO密集型任务：由于IO操作速度远低于CPU速度，那么在运行这类任务时，CPU绝大多数时间处于空闲状态，那么线程池可以配置尽量多些的线程，以提高CPU利用率；

对于混合型任务：可以拆分为CPU密集型任务和IO密集型任务，当这两类任务执行时间相差无几时，通过拆分再执行的吞吐率高于串行执行的吞吐率，但若这两类任务执行时间有数据级的差距，那么没有拆分的意义。