【JDK7源码阅读计划】线程池
2017-05-16 本文已影响0人
沐星河
梦想在没有实现之前，不必对他人讲。
先从全局看问题总是没错的，线程池的继承体系：
Executors 是一个用来生产线程池的静态工厂，可以通过该类生产ExecutorService、ScheduledExecutorService等对象。
在 Executors 这个类里面，定义了这么几种常用的线程池：
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>());
}
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
return new DelegatedScheduledExecutorService
(new ScheduledThreadPoolExecutor(1));
}
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
return new ScheduledThreadPoolExecutor(corePoolSize);
}
这几种线程池都构造了ThreadPoolExecutor类，只是参数不同，所以看一下这个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.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
ThreadPoolExecutor参数描述如下：
-
corePoolSize线程池核心线程数。当提交一个任务时，线程池会新创建一个新线程执行任务，直到线程数达到corePoolSize；之后继续提交的任务会被保存到阻塞队列中。 -
maximumPoolSize线程池最大线程数。这个参数只有在队列有界的情况下才有效。当前阻塞队列满了的情况下，继续提交任务时，则会继续创建新的线程执行任务，直到线程数达到maximumPoolSize。之后再提交任务，会执行拒绝策略。 -
keepAliveTime空闲队列存活时间。大于corePoolSize的空闲线程在该时间之后会被销毁 -
unitkeepAliveTime 的单位 -
workQueue阻塞队列，一般有如下几种阻塞队列
- ArrayBlockingQueue：基于数组的有界阻塞队列
- inkedBlockingQuene：基于队列的无界阻塞队列
- SynchronousQuene：不实际存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，反之亦然。如果使用该队列，提交的任务不会保存，而总是将新任务提交给线程执行，如果没有空闲线程，则尝试创建新的线程，如果线程已达最大值，则执行拒绝策略。
- priorityBlockingQuene：具有优先级的无界阻塞队列
-
threadFactory线程工厂 -
handler拒绝策略，当队列已满，且没有空闲线程时，会执行一种拒绝策略，JDK一共有四种拒绝策略
- AbortPolicy：直接抛出异常
- CallerRunsPolicy ：在调用者线程中运行任务
- DiscardOldestPolicy： 丢弃最早的一个请求，再次提交该任务
- DiscardPolicy: 直接丢弃，不做任何处理
结合之前的代码可以看到，当corePoolSize 等于maximumPoolSize 时，构造的就是newFixedThreadPool，这两个都为1 时，构造的是newSingleThreadExecutor。newCachedThreadPool线程池在没有任务执行时，数量为0，其数量会动态变化，最大值为Integer.MAX_VALUE`
ScheduledThreadPoolExecutor 继承了ThreadPoolExecutor,构造方法：
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
new DelayedWorkQueue());
}
ScheduledThreadPoolExecutor增加了一些定时任务的功能，这里使用到了DelayedWorkQueue，这个队列也很有意思，模拟了二叉查找树的性质，用来存放有序的计划任务。
主要方法如下：
//在指定的时间后，对任务进行一次调度
public ScheduledFuture<?> schedule(Runnable command,
long delay, TimeUnit unit);
//对任务进行周期性调度，以开始时间计算，周期性调度
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
long initialDelay,
long period,
TimeUnit unit);
//对任务进行周期性调度，以结束时间计算，经过延迟后，才进行下一次
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
long initialDelay,
long delay,
TimeUnit unit);
那么在线程池中的线程是如何调度的，线程池的原理是什么呢？
先看一下线程池的状态表示：
//这个原子类非常强大，其中的高3为表示线程池状态，后29位表示线程数
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3; //29
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//高3位为111，表示线程池能接受新任务，并且可以运行队列中的任务
private static final int RUNNING = -1 << COUNT_BITS;
//高3位000，表示线程池不再接受新任务，但可以处理队列中的任务
private static final int SHUTDOWN = 0 << COUNT_BITS;
//高3为001，表示线程池不再接受新任务，不再执行队列中的任务，而且要中断正在处理的任务
private static final int STOP = 1 << COUNT_BITS;
//高3位010，表示线程池位为空，准备关闭
private static final int TIDYING = 2 << COUNT_BITS;
//高3位011，表示线程池已关闭
private static final int TERMINATED = 3 << COUNT_BITS;
//获取高3位
private static int runStateOf(int c) { return c & ~CAPACITY; }
//获取低29位
private static int workerCountOf(int c) { return c & CAPACITY; }
//将高3位，低29位保存在一个int里
private static int ctlOf(int rs, int wc) { return rs | wc; }
接下来分析线程池的调度代码，当我们用线程池执行一个任务的时候，会执行以下方法。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//获取ctl值，上面的分析知道，这个值包含了高3位的线程池状态和低29位的线程池数量
int c = ctl.get();
//拿到线程数量和核心线程数比较
if (workerCountOf(c) < corePoolSize) {
// 如果当前线程数量< 核心线程数，则执行addWorker 方法，这个方法会新建线程并执行任务
if (addWorker(command, true))
return;
//如果执行失败，再拿一次ctl的值
c = ctl.get();
}
// 当线程数大于核心线程，或上边任务添加失败时
// 在线程池可用的时候，会将任务添加到阻塞队列中
if (isRunning(c) && workQueue.offer(command)) {
// 再次确认线程池状态，若线程池停止了，将任务删除，并执行拒绝策略
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
//如果线程数量为0，则放入一个空任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//如果队列无法放入，则再新建线程执行任务，如果失败，执行 拒接策略
// 这里就是从core 到 max 的扩展
else if (!addWorker(command, false))
reject(command);
}
下面看一下addWorker方法
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
// 获取线程池状态
int rs = runStateOf(c);
// 如果线程池不在运行状态，则不再处理提交的任务，直接返回 , 但可以继续执行队列中已有的任务
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//这里的死循环是为了CAS 线程数量，直到成功之后跳出外层循环
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 {
final ReentrantLock mainLock = this.mainLock;
// 用传进来的任务构造一个worker ，该类继承了AQS，实现了Runnable
w = new Worker(firstTask);
// 获取worker中创建的线程
final Thread t = w.thread;
if (t != null) {
//加锁 ,HashSet线程不安全
mainLock.lock();
try {
int c = ctl.get();
int rs = runStateOf(c);
// 检测线程池状态
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();
}
//添加成功之后，启动worker线程
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
//返回值标识线程是否启动
return workerStarted;
}
看一下线程是怎么启动的:
// worker类
Worker(Runnable firstTask) {
//在运行之前不允许中断
setState(-1);
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
线程启动执行的是runWorker方法
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
//由于在worker构造方法中抑制了中断，这里解除抑制
w.unlock(); // allow interrupts
//默认为true，说明发生了异常
boolean completedAbruptly = true;
try {
//先执行传进来的任务，之后从队列获取任务执行
while (task != null || (task = getTask()) != null) {
w.lock();
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();
}
}
// 如果一切正常，置为false , 清理时会做判断
completedAbruptly = false;
} finally {
//清理工作，同时 任务如果有异常，会通过这个方法擦屁股
processWorkerExit(w, completedAbruptly);
}
}
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
boolean timed; // Are workers subject to culling?
// 两种情况：
// 1.RUNING状态
// 2.SHUTDOWN状态，但队列中还有任务需要执行
for (;;) {
int wc = workerCountOf(c);
timed = allowCoreThreadTimeOut || wc > corePoolSize;
if (wc <= maximumPoolSize && ! (timedOut && timed))
break;
// 执行到这里说明线程已超核心线程数并且超时，这时返回null回收线程
if (compareAndDecrementWorkerCount(c))
return null;
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
try {
//如果核心线程允许超时，或者线程数已达到核心线程数，则执行poll
//poll方法在规定时间内没返回会返回null，在下一轮循环的时候，会返回null,线程会被销毁
// 否则，执行take方法，该方法会阻塞直到队列中有任务，所以当线程数在核心线程数以下的线程不会被销毁
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
最后看一下runWorker中的清理工作：
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//如果非正常结束，将线程数减一
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
//从线程池中移出异常和超时的线程
workers.remove(w);
} finally {
mainLock.unlock();
}
// 尝试关闭线程池
tryTerminate();
int c = ctl.get();
//线程池状态在RUNNING或SHUTDOWN时
if (runStateLessThan(c, STOP)) {
// 线程正常结束
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
//如果线程为0 但是队列中还有任务要执行
if (min == 0 && ! workQueue.isEmpty())
min = 1;
//线程数量满足条件，直接返回
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//新建空的任务，假如队列中有任务的话，这里保证能执行
//如果线程是因为异常退出的，这里进行补充
addWorker(null, false);
}
}
final void tryTerminate() {
for (;;) {
int c = ctl.get();
// 线程池正在运行时
// 线程池是SHUTDOWN状态，但是队列还有任务时
// 线程池已经准备停止时 直接返回
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
//下面的代码说明线程池真的需要关闭了
//如果线程数量不为0，说明需要将线程中断，这里只中断一个线程就可以（为啥呢？）
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
//执行关闭操作
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 使用 CAS 设置状态位
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
到这里，线程池的基本原理基本能明白一二吧...
