-
一、前奏
-
二、开始
-
三、分析
-
3-1、execute(Runnable command)
-
3-2、 addWorker(Runnable firstTask, boolean core)
-
3-3、Worker
-
3-4、runWorker(Worker w)
-
3-5、getTask()
-
四、问题剖析
-
4-1、添加的一个任务是怎么运行的？
-
4-2、任务丢到队列，怎么取出来呢？
-
4-3、过了时间怎么销毁线程？
-
4-4、怎么拒绝的？
-
4-5、线程池，这个池是什么？ 线程怎么放进去？
一、前奏
有了上一篇博文的学习，相信你对于线程池的使用这块已经不在存在什么问题了，日常开发和面试也都足够了——线程池最优使用策略【Java线程池学习一】
但随着时间的推移在闲下来的时候我突然想，当任务进入了队列之后是怎么取出来的呢？然后列举了几个问题
-
添加的一个任务是怎么运行的？ -
任务丢到了队列，怎么取出来呢？ -
过了时间怎么销毁线程？ -
怎么拒绝的？ -
线程池，这个池是什么？ 线程怎么放进去？
毫无疑问想要解决上面的问题，那只有研究源码，下面我们就来看下 ThreadPoolExecutor 的源码，此次目的就是解决上面的问题，先对线程池的核心工作原理进行理解，后面我们再来对线程池来一个全面的解读。
二、开始
public static void main(String[] args) throws InterruptedException {
ThreadPoolExecutor executor = new ThreadPoolExecutor(1,2, 1, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));
executor.execute(()->{
try {
Thread.sleep(1000 * 6);
System.out.println("task-1 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
} catch (InterruptedException e) {
e.printStackTrace();
}
});
executor.execute(()->{
System.out.println("task-2 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
});
executor.execute(()->{
System.out.println("task-3 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
});
Thread.sleep(1000 * 10);
executor.execute(()->{
System.out.println("task-4 " + Thread.currentThread().getName()+ " " + Thread.currentThread().hashCode());
});
}
上面的代码案例很简单，现在我们用上一篇博文的知识来解读一下这段代码：
-
task-1 进入线程池，这时候线程池里面还没有线程，所以会启一个线程去执行【thread-1】 -
task-2 进入线程池，虽然线程池里面有一个【thread-1】，但是它被上一个任务阻塞着，所以 task-2进入队列 -
task-3 进入线程池，这时候队列满了，【thread-1】还在阻塞中，所以会开启一个新的线程去执行【thread-2】 -
task-3 执行完毕后，队列中还有一个 task-2，所以【thread-2】会去执行 task-2 -
执行完 task-2 之后，我们暂停添加任务，【thread-2】会在等待 1s 之后就销毁了 -
这时候 task-1执行完毕， 过了一会我们又添加了一个 task-4 到线程池，【thread-1】是长期存活的它会去执行 task-4
打印结果如下：
task-3 pool-1-thread-2 448689266
task-2 pool-1-thread-2 448689266
task-1 pool-1-thread-1 1721383753
task-4 pool-1-thread-1 1721383753
三、分析
注：ThreadPoolExecutor 的源码里面用到了 与或非 左移右移 运算，理解起来比较困难，但又不是业务的主流程，所以我在后面进行简化。
3-1、execute(Runnable command)
上面的案例代码我们只是调用了线程池的 execute 方法 ，所以我们直接来看看这个方法。
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/**
* workerCountOf(c) 就是返回的当前线程池存在的线程数
*
* 如果小于核心线程数就进行 addWorker ，把任务添加进去，添加成功就结束（注这个addWorker并不是加入队列）
*/
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
/**
* 当前线程池没有在跑，并成功把当前任务放入队列中
*
* 插入成功后会检查一下当前线程池的运行情况，如果isRunning返回false说明线程池已经超负荷运行了 就删除任务 并拒绝任务
*
* 如果上一步没有问题，就再判断一下当然线程池是否有在运行的线程，没有就开启一个，下面我们会详细说明【addWorker】
*/
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);
}
-
offer 方法是在队列允许的范围内把数据插入进去，如果容量不足就返回false -
线程池的实时性很高，所以可以看到它的源码里面做了很多补偿的机制，比如任务插入到队列中去后还会再判断线程池的状态是否正常。（如果在插入成功的一瞬间线程池已经终止了就会出问题）
3-2、 addWorker(Runnable firstTask, boolean core)
-
firstTask 就是当前任务，当然它可以为null，如果是null就是让当前线程去执行队列中其它任务（后面会看到） -
core 判断线程的数量是在 核心线程数量还是最大线程数量的范围 （true 就是核心线程数量范围内）
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;
for (;;) {
// wc 为当前线程池的线程数
int wc = workerCountOf(c);
// 如果线程数大于最大容量 或 大于规定容量就返回 false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 使用CAS去让当前线程数+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 是private final修饰的内部类，实现了 Runnable接口，里面有【当前任务和当前线程】下面详解
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());
/**
* 判断当前任务是否符合加入到任务中的条件
*
* rs 是一个很大的负数 可以理解规定能最大创建的线程数【536870911】 所以不用考虑，不会创建那么多
* SHUTDOWN 是个常数 0
*/
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// 判断当前线程是否可以被 开启/start
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.isAlive() 可以参看 https://www.javatpoint.com/java-thread-isalive-method
3-3、Worker
现在我们来看一下Worker这个内部类，上面我们也说了Worker 实现了Runnable，并且在最后的一步中调用了 start方法。
它继承了AbstractQueuedSynchronizer 实现了 Runnable，但代码很简单，定义了三个参数，代码比较简单直接看。
-
thread 执行当前任务的线程 -
firstTask 当前任务 -
completedTasks 该线程执行成功的任务数 （会统计整个线程池执行成功的任务数）
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable {
private static final long serialVersionUID = 6138294804551838833L;
final Thread thread;
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
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) {
}
}
}
}
我们单独来看一下它的构造方法，任务是由外面传递过来的，线程是每次创建一个新的，上面我们在创建的时候没有指定ThreadFactory，则用的默认的 Executors.DefaultThreadFactory
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
3-4、runWorker(Worker w)
3-2 里面我们调用了 worker 的 start方法，后续肯定就会调用 run 方法了，也就是 runWorker 方法
public void run() {
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 里面没有任何实现是空方法，可以留给后续扩展。 -
我们以为当前线程只是去执行当前的 任务/task， 但其实不是 task != null || (task = getTask()) != null 它先去执行自己的任务，如果自己的任务执行完了，就会执行 getTask 从队列中拿任务直到队列中没有任务为止。 -
我们在上面也看到调用 addWorker方法的时候 任务传递的是 null，其实就是让它去执行队列中的任务。
3-5、getTask()
该方法如同它自己的名【获取任务】，它的作用就是从队列中获取任务。
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 特殊情况下校验队列是不是为空
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
/**
* allowCoreThreadTimeOut 是否允许核心线程超时 默认 false
* wc > corePoolSize 当前活跃线程是否大于核心线程数
*
* 其实就是相当于只有超过核心线程数的线程才设置超时时间
*/
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
/**
* 如果当前线程数 > 最大线程数 就退出（销毁当前线程）
* 或 当前线程是设置了超时的，并且已经超时了 就退出（销毁当前线程）
*/
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
/**
* 从队列中获取数据
*
* poll 等待固定时间
* take 一直阻塞，直到获取到了任务数据
*/
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
take方法 一直阻塞，直到获取到了任务数据
四、问题剖析
4-1、添加的一个任务是怎么运行的？
-
黑色的线是正常添加一个任务的流程 -
红色的线是线程组里面的线程一直阻塞获取队列中的数据
4-2、任务丢到队列，怎么取出来呢？
下面是截取 getTask() 里面的一段代码
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
简单理解就是 非核心线程调用 poll 方法，核心线程调用 take 方法，take方法是阻塞的，获取不到数据会一直等着。
4-3、过了时间怎么销毁线程？
咱们把 getTask 方法复制过来，删掉多余的部分
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
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;
}
}
}
我们知道超过核心线程的部分会调用 poll 方法，如果超过时间没有获取到对应的任务： timedOut = true
如果 timedOut = true，那上面的if判断就会进去，当前死循环就结束了，线程也就正常结束。
4-4、怎么拒绝的？
拒绝这个就很简单了，在excute方法的讲解里面已经说了。
4-5、线程池，这个池是什么？ 线程怎么放进去？
再来回顾一下是怎么创建线程的，在进入 addWorker 方法的时候会进行一系列的判断，当判断通过后就会创建一个 w = new Worker(firstTask);
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
找到这个 Worker 的构造方法再来看一下，前面我们也说了默认的线程工厂是Executors.DefaultThreadFactory，来看看它的 newThread 方法
DefaultThreadFactory
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
public Thread newThread(Runnable r) {
Thread t = new Thread(group, r,
namePrefix + threadNumber.getAndIncrement(),
0);
if (t.isDaemon())
t.setDaemon(false);
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
可以看到其实每次都是 new Thread，但是它的 ThreadGroup 是同一个。
线程启动的时候是调用的 Thread.start() 方法，隐藏其它方法，里面有一个 把当前线程添加到线程组里面。
public synchronized void start() {
// ...
group.add(this);
// ...
}
ThreadGroup.add
int nthreads;
Thread threads[];
void add(Thread t) {
synchronized (this) {
if (destroyed) {
throw new IllegalThreadStateException();
}
if (threads == null) {
threads = new Thread[4];
} else if (nthreads == threads.length) {
threads = Arrays.copyOf(threads, nthreads * 2);
}
threads[nthreads] = t;
nthreads++;
nUnstartedThreads--;
}
}
原文始发于微信公众号（小道仙97）：Java线程池运行原理，线程池源码解读【Java线程池学习二】
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/204478.html
