newSingleThreadExecutor源码解析&&JDK提供线程池ThreadPoolExecutor执行任务流程解析
- 测试代码
- 核心代码解析
- ThreadPoolExecutor内部类：Worker解析
- 调试台线程切换到执行异步任务的线程
- ThreadPoolExecutor：getTask()
- LinkedBlockingQueue：take()
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
初始化的线程池中只有一个线程，如果该线程异常结束，会重新创建一个新的线程继续执行任务，唯一的线程可以保证所提交任务的顺序执行
由于使用了无界队列, 所以SingleThreadPool永远不会拒绝, 即饱和策略失效
测试代码
public class ExecutorsTest {
public static void main(String[] args) {
ExecutorService threadExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNamePrefix("thread_zs_").build());
AtomicInteger atomicInteger = new AtomicInteger(1);
for (int i = 0; i < 3; i++) {
threadExecutor.submit(() -> {
task(10L, TimeUnit.SECONDS, atomicInteger);
});
}
}
public static void task(Long millisecond, TimeUnit timeUnit, AtomicInteger atomicInteger) {
int andIncrement = atomicInteger.getAndIncrement();
System.out.println(Thread.currentThread().getName() + "---执行任务" + andIncrement +"-----start");
try {
Thread.sleep(timeUnit.toMillis(millisecond));
System.out.println(Thread.currentThread().getName() + "---执行任务" + andIncrement+"-----end");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
核心代码解析
execute()
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) { // 0 < 1 true workerCountOf(当前创建线程数) -------1
if (addWorker(command, true)) // 创建工作线程 ------2
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) { // isRunning(c) false
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);
}
创建工作线程并执行任务
private boolean addWorker(Runnable firstTask, boolean core) { // core = true
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 校验队列是否为空 ---------3 不成立
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c); // wc = 0
if (wc >= CAPACITY || // 当前工作线程数是否大于最大线程数容量
wc >= (core ? corePoolSize : maximumPoolSize)) // 当前线程数与核心线程数比较 0 >= 1 false
return false;
if (compareAndIncrementWorkerCount(c)) // 当前线程数+1 --------4 原来ctl是-536870912，cas之后+1
变为 -536870911
break retry; // 跳出最外层循环 ----------5
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); ----------6 创建线程
final Thread t = w.thread;
if (t != null) { // 线程不为空，加锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get()); // -------10 ctl: -536870911 rs: -536870912
// rs 为当前线程池状态，线程获取锁后重新检查线程池状态
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable // 线程池状态正常
throw new IllegalThreadStateException();
workers.add(w); // 将该线程添加到工作线程中，workers为hashset
int s = workers.size(); // 1
if (s > largestPoolSize) // 1 > 0
largestPoolSize = s;
workerAdded = true; // --------11 工作线程添加成功
}
} finally {
mainLock.unlock(); // ----------12 解锁
}
if (workerAdded) {
t.start(); // --------13 执行任务，t是Worker的实例变量 注意这里打断点的调试会有所不同，具体的大家看看后面的解析
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
ThreadPoolExecutor内部类：Worker解析
创建线程并设置线程任务，把自身引用当做Runnable放入到创建出来的线程中
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
final Thread thread;
Worker(Runnable firstTask) { // -------------7
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this); // 将自身引用当做runnable放入到创建出来的线程中 -----8
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
入参target就是新创建的woker，newThread().start() 也就是执行new Worker().run()，最终执行的是runWorker(this)
@Override
public Thread newThread(final Runnable target) {
return new Thread(this.group, target, this.namePrefix + "-" + this.count.incrementAndGet()); // -----9
}
调试台线程切换到执行异步任务的线程
在上面断点到第13步（957行）时，我们观察到调试台目前有4个线程，Finalizer与Reference Handler可能是处理软引用使用的吧，具体也不去关心啦，有兴趣的小伙伴可以去查查看。
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fROO3LIh-1668837082985)(/Users/zhengshuang/Library/Application Support/typora-user-images/image-20221118161459035.png)]](https://img-blog.csdnimg.cn/1310d2a83cf247098a1ca12095f40a0c.png)
调试台 Step Over后，发现多了一个名为thread_zs_0的线程，我们切换到该线程，继续代码调试
注意：原来我们调试的线程，是走main 方法中threadExecutor.submit() 用来提交任务的，提交时会遇到各种校验，策略的判断，真正执行线程任务的是线程池里边的线程，不是当前主线程
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7jvyzHdC-1668837082986)(/Users/zhengshuang/Library/Application Support/typora-user-images/image-20221118162302649.png)]](https://img-blog.csdnimg.cn/3592272438ed440ebb0bebef02c69465.png)
切换线程后，执行的是工作线程
@Override
public void run() {
if (target != null) {
target.run(); // -------14 执行thread 的run方法，这里的target，就是前面创建的Worker，Worker实现了Runnable接口，在构造方法中调用init()完成对target的赋值
}
}
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
public void run() {
runWorker(this); // ---------15 调用的是Worker中的runWorker
}
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread(); // 获取当前线程
Runnable task = w.firstTask; // 创建worker时传参的runnable任务
w.firstTask = null; // gc
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// runnable不为空
while (task != null || (task = getTask()) != null) { // ---------21 第二次循环task==null，走getTask(),在下面进行解析
// ---------31 getTask()获取到任务，往下走执行完任务又回到getTask(); 32 解析会在getTask()方法的workQueue.take()中
w.lock();
// 显然下面的if一定是false，线程池状态不是stop
// 下面这段if判断的主要目前就是，判断如果线程池状态是stop的，就看看执行当前任务的线程是否被打断，如果就一定确保当前线程被打断
if (-----------1 start ----------------
( -----------2 start --------------
runStateAtLeast(ctl.get(), STOP) // 判断当前线程池状态是否是停止的
||
// 走到这说明 || 之前的是false 当前线程的状态是正常的
(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP)) // 当前线程池是正常的，需要判断线程是否被打断，需要重新检查线程池是否停止（Thread.interrupted()判断线程是否被打断，同时会清除打断状态，也就是说调用这个方法后，线程的的打断状态一定是false）
----------2 end ------------------)
&&
// 走到这 说明 && 之前是true，也就是说当前的线程池一定是stop状态
!wt.isInterrupted() // 如果起前面没有执行Thread.interrupted()这里可能会返回false，如果是false就取反，一定要把线程打断
-------------1 end --------------)
wt.interrupt();
try {
beforeExecute(wt, task); // ---------16 空方法实现
Throwable thrown = null;
try {
task.run(); // --------17 到这里才是真正执行run方法的地方，task就是创建worker传入的runnable
} 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); // ----------18 空方法实现，此时另外两个任务已经加入到任务队列中
}
} finally {
task = null; // --------19 为能够执行到while循环条件的第二个做准备
w.completedTasks++; // 完成的任务数统计，被voltiale修饰，多线程课件
w.unlock(); // 解锁
} // -------20 准备进入下一循环回到上面的while
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
ThreadPoolExecutor：getTask()
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c); // 获取线程池状态 --------- 22
// Check if queue empty only if necessary.
// 判断线程池状态>=shutdown(000)，shutdown不会接收新任务，但会处理阻塞队列剩余的任务
// >= 000,就接着判断是否>=STOP(001) ,stop会中断正在执行的任务，并抛弃阻塞队列的任务
// 如果这是当前线程池状态只满足shutdown，并且线程池中任务队列为空，就减少当前工作线程的数量
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount(); // 通过cas改变ctl的值
return null;
}
int wc = workerCountOf(c); // 获取当前工作线程数量 1
// allowCoreThreadTimeOut false
// wc 1 > corePoolSize 1 false
// timed决定工作线程是否被淘汰 false
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // ----------- 23
// 1 > 1 false || fasle & false
if
(
(wc > maximumPoolSize || (timed && timedOut)) // false
&&
(wc > 1 || workQueue.isEmpty())
) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ? // false
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take(); // 走的是这个 --------------24 从任务队列中取值 在下面进行分析
if (r != null) // --------30 获取得到runnable不为空
return r; // -------31 回到方法调用处
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
LinkedBlockingQueue：take()
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count; // ------------25 当前有两个任务，为什么是两个，因为我们在main主线程中循环了3遍，其中第一个任务也就是我们现在分析的这个一直打断点卡着这个线程，其他两个任务就被放入任务队列里，等待调度执行
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly(); // 加了等待获取锁中可被打断的锁
try {
while (count.get() == 0) { // 此时任务队列有两个任务
notEmpty.await(); // 阻塞等待
}
x = dequeue(); // ---------26 获取任务队列中的任务
c = count.getAndDecrement(); // ----------28 先获取当前任务队列保存的任务数后在-1，该count是局部变量，只在当前栈帧有效
if (c > 1) // 2 > 1 true // 32 准备获取第三个任务，c此时为0
notEmpty.signal(); // --------27 工作队列中取完一个任务后还有任务，看看有没有其他工作现在执行完了，但是在前几行判断任务队列为空时，被notEmpty.await()阻塞住了，但是这里有个问题,就是前面while(count.get()==0){notEmpty.await()},既然notEmpty.signal()会去condition中随机按照队列顺序去唤醒第一个节点，但是在notEmpty.await()处阻塞被唤醒的线程，为防止虚假唤醒，会接着走while循环去判断，但是count在该线程调用take()，就已经创建好了这个栈帧，count也是在创建的时候就被赋值，那么count的值将会是一直不变的，那唤醒后岂不是又进入阻塞了？
// 但是其实不会的，因为final AtomicInteger count = this.count; 传递的是引用，java只有值传递，但是遇到基本类型确实传递的是基本类型的值，但是遇到引用类型，传递的值是引用
} finally {
takeLock.unlock();
}
if (c == capacity) // ---------28 c的取值是count.getAndDecrement(),先取值后自减，也就是说如果当前c的大小与任务队列容量大小一样，则有剩余空间让阻塞的put操作往下执行，使用的条件变量(new ReentrantLock().newCondition())是notFull
signalNotFull();
return x; // ----------29 take()方法调用完毕，弹出该栈帧，回到方法被调用处
}
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
E x = first.item; // --------27 队列中head的item为null，从第二个node开始才会真正存放runnable
first.item = null;
return x;
}
}
以上便是JDK提供线程池ExecutorService中newSingleThreadExecutor的执行流程以及部分代码解析，如有误解，请在评论区指出，谢谢
ExecutorService提供的newFixedThreadPool、newCachedThreadPool的源码解析也会在近期发布，敬请期待！
