在说线程池工作流程之前,先说清楚线程池中字段的含义以及作用.
说明,ctl是真个线程池状态的核心,通过巧妙的方式保存了两个状态,
1.所有有效线程的数量workerCount简写wc
2.线程池的状态runStatus简写rs
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
线程池的几个状态
1.Running状态,表示线程池是运行状态,能够接受新的任务,也能处理阻塞队列中的任务.
2.Shutdown状态,表示关闭状态,不能接受新的任务,但是能处理阻塞队列中的任务.当线程
池处于Running状态,调用Shutdown方法可以使线程池进入该状态
3.Stop状态,表示停止状态,不能接受新任务,也不能处理阻塞队列中的任务.当线程池处于
Running状态或者Shutdown状态时,可以调用Stop方法,可以使线程池进入该状态.
4.Tidying状态,表示清理状态,当所有任务都终止了,wc为 0时时,线程池会调用Terminated
方法让线程池进入Terminated状态,如果此后线程池wc为0并且阻塞队列中任务为0,线程池就会进入此
状态.当线程池为Stop状态并且wc为0时也进入此状态.
5.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;
通过ctl计算两个状态
1.计算线程池的状态
2.计算当前线程的数量
反之也可以通过线程池的状态和房钱线程的数量计算ctl
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; }
阻塞队列,用来存放任务的,当一个新的任务提交到线程池之后,线程池会根据当前线程池中正在运行的
线程数量来对该任务做处理,主要有以下的处理方式.
1.如果当前线程池中正在运行的线程数小于核心线程数,那么线程池总是倾向于创建一个新的线程去
执行,而不是把任务放入任务队列中等待.
2.如果当前线程池中正在运行的线程数大于等于核心线程数,那么线程池总是倾向于把任务放入任务
队列中,而不是创建新线程去执行.
3.如果当前线程池中正在运行的线程数大于等于核心线程数,并且任务队列已满的情况下,线程池会
去判断当前池中的线程数是否已经等于了最大线程数maximumPoolSize,如果已经等于了最大线程数,
那么就会执行拒绝策略,否则会创建一个线程去执行该任务.
常见使用的队列有三种:
1.SynchronousQueue 同步队列,(任务队列中只有一个任务,只有消费了该任务才能继续提交任务)
2.无界队列,如LinkedBlockingQueue,(可以一直往任务队列中插入任务)
3.有界队列,如ArrayBlockingQueue
private final BlockingQueue<Runnable> workQueue;
工作线程组,消费阻塞队列中的任务.必须持有mainLock才能够进入.
Worker类中包含了一个线程,真正处理队列中的任务,之后会详说Worker类
private final HashSet<Worker> workers = new HashSet<Worker>();
拒绝策略,线程池执行拒绝策略有两种情况
1.当线程数已经达到最大线程数并且阻塞队列中无法放入任务的时候,执行拒绝策略.
2.当线程池处于Stop状态.
jdk中实现了4中拒绝策略:
1.AbortPolicy 直接抛出异常(JDK线程池实现中默认方式).
2.CallerRunsPolicy 将任务放在执行该线程的方法中执行.
3.DiscardPolicy 不执行提交的任务.
4.DiscardOldestPolicy 如果线程池已经关闭了则使用DiscardPolicy方式,如果没
有关闭则抛弃队列中的head任务,将这个新提交的任务放入队列.
private volatile RejectedExecutionHandler handler;
keepAliveTime表示当空闲线程处于等待状态的超时时间,
当最大线程数大于核心线程数的时候,并且allowCoreThreadTimeOut被设置为false,多出
了的非核心线程数等待时间如果到达超时时间则该线程被回收
private volatile long keepAliveTime;
默认情况下线程池是不会回收核心线程的,当allowCoreThreadTimeOut为ture的时候,当线程空闲时
间大于超时时间的时候.线程被回收
private volatile boolean allowCoreThreadTimeOut;
核心线程数
private volatile int corePoolSize;
最大线程数
private volatile int maximumPoolSize;
上面说完了线程池中主要的一些概念,下面说下线程池的调度原理.
当我们向线程池提交了一个任务的时,可以通过submit或者execute方法来调用实现,两者的区别是一
个当submit执行完成之后可以返回结果,execute不能返回结果,但内部实现都是利用execute来完成
的.所以以下都通过分析execute来了解调度的原理.以下基于1.8的源码来分析:
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的值,ctl中包含了线程池的状态和运行的线程数
int c = ctl.get();
// 当工作线程数小于和核心线程数,尝试开启一个新的线程,把传入的command作为第一个
任务.
if (workerCountOf(c) < corePoolSize) {
//
if (addWorker(command, true))
return;
c = ctl.get();
}
// 如果当工作线程数大于等于核心线程数,线程池处于Running状态,并且将任务插入队列
成功
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 重新获取ctl,如果线程池的状态不是running状态,那么就把刚刚加入的任务移除掉,并
且执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
//如果发现当前线程池中的线程数为0,那么就设置最大线程数为maximumPoolSize,并且创
建一个线程,但是不start该线程.
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果能到达这里有两种情况:
// 1.线程池不是running状态,执行拒绝策略
// 2.线程池是running状态,但是队列已满不能加入任务队列,线程池会判断是否当前线
程池中的线程是否大于maximumPoolSize,如果大于maximumPoolSize,则执行拒绝策略,否则创建一
个线程执行该任务.
else if (!addWorker(command, false))
reject(command);
}
总结下线程池的执行过程:
1: 如果线程池内的有效线程数少于核心线程数 corePoolSize, 那么就创建并启动一个线程来执行新
提交的任务.
2: 如果线程池内的有效线程数达到了核心线程数 corePoolSize, 并且线程池内的阻塞队列未满, 那
么就将新提交的任务加入到该阻塞队列中.
3: 如果线程池内的有效线程数达到了核心线程数 corePoolSize 但却小于最大线程数 maximumPoolSize,
并且线程池内的阻塞队列已满, 那么就创建并启动一个线程来执行新提交的任务.
4: 如果线程池内的有效线程数达到了最大线程数 maximumPoolSize, 并且线程池内的阻塞队列已满,
那么就让 RejectedExecutionHandler 根据它的拒绝策略来处理该任务, 默认的处理方式是直接
抛异常.
在上面的调度流程分析中我们发现execute中大量了使用了addWorker方法,并且三次传递的参数都是不同的,我们分析下:
addWorker这个方法:
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 如果线程处于以下几个状态,将直接返回
// 1.处于STOP,TYDING, SHUTDOWN状态
// 2.处于SHUTDOWN状态并且firstTask!=null
// 3.处于SHUTDOWN状态,并且任务队列为空.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
// 如果线程池内的线程数量是以下的几种情况,将直接返回:
// 1.如果线程池内的线程数大于理论上的最大容量CAPACITY
// 2.当core为true时,线程池内线程数量大于corePoolSize核心线程数.
// 3.当core为false时,线程池内线程数量大于maximumPoolSize最大线程数
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 将线程池内的线程数加1
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 {
// 创建Worker类,将传入的command作为第一个任务,在创建Worker的时候,内部会自
动的创建一个线程.
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();
// 将worker放入到这个工作线程包装类的集合中.
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 增加成功,开启线程.
if (workerAdded) {
// 开启线程执行任务
// 说明: t指向的是w.thread,也就是worker内的线程.我们看下worker
内的thread是如何创建的
/*
* Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
*/
// thread在创建的时候把this传入了.也就是说当t.start开启线程之后,
会执行该线程的包装类worker的run方法.
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
现在已经分析到开启线程了,一般线程在执行完run方法会自动的销毁改线程,那么下面分析线程池是如何复用线程的.
之前说到开启线程之后,执行的是worker内的run方法,那我们线程追到worker类中的run方法.
public void run() {
runWorker(this);
}
我们发现run方法委托给runWoker方法处理.继续跟踪到runWorker方法中.
final void runWorker(Worker w) {
// 获取当前线程
Thread wt = Thread.currentThread();
// 获取第一个任务后,把第一个任务置为null
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// 判断如果第一个任务不为空,或者getTask不为空(getTask是从任务队列中阻塞的获
取任务,保证线程不会被回收)
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 {
// 在这可以看到我们在手动执行任务
// 这个逻辑大概是:
// 1.开启一个线程执行run方法.
// 2.在run方法中不断获取任务队列中的方法,显示的调用.这样我
们就能复用线程了.
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);
}
}
下面我们在看下getTask方法.
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
// 一个死循环
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
// 阻塞的方式消费任务.
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
以上是线程池的大概流程,还有很多地方没有说明,比如一些异常情况的处理,多线程下的状态判断等等.希望大家能够多提宝贵的建议或意见, 互相交流学习!
