首先看一段线程池的使用代码
public class FixedThreadPoolDemo {
static ExecutorService executorService = Executors.newFixedThreadPool(3);
public static void main(String[] args) {
for (int i = 0; i < 100; i++) {
executorService.execute(() -> {
System.out.println(Thread.currentThread().getName() + "-> 执行");
});
}
// 关闭线程池
executorService.shutdown();
}
}
一语概括线程池就是启动线程去执行任务的run（）方法，创建的worker是真实的线程，这段代码就是任务的run方法
image.png
查看线程池的execute方法如下
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
//第一步，获取ctl
int c = ctl.get();
//检查当前线程数是否达到核心线程数的限制，注意线程本身是不区分核心还是非核心，后面会进一步验证
if (workerCountOf(c) < corePoolSize) {
//如果核心线程数未达到，会直接添加一个核心线程，也就是说在线程池刚启动预热阶段，
//提交任务后，会优先启动核心线程处理
if (addWorker(command, true))
return;
//如果添加任务失败，刷新ctl，进入下一步
c = ctl.get();
}
//检查线程池是否是运行状态，然后将任务添加到等待队列，注意offer是不会阻塞的
if (isRunning(c) && workQueue.offer(command)) {
//任务成功添加到等待队列，再次刷新ctl
int recheck = ctl.get();
//如果线程池不是运行状态，则将刚添加的任务从队列移除并执行拒绝策略
if (! isRunning(recheck) && remove(command))
reject(command);
//判断当前线程数量，如果线程数量为0，则添加一个非核心线程，并且不指定首次执行任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//添加非核心线程，指定首次执行任务，如果添加失败，执行异常策略
else if (!addWorker(command, false))
reject(command);
}
调用了worker方法如下
private boolean addWorker(Runnable firstTask, boolean core) {
//相当于goto，虽然不建议滥用，但这里使用又觉得没一点问题
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
//如果线程池的状态到了SHUTDOWN或者之上的状态时候，只有一种情况还需要继续添加线程，
//那就是线程池已经SHUTDOWN，但是队列中还有任务在排队,而且不接受新任务（所以firstTask必须为null）
//这里还继续添加线程的初衷是，加快执行等待队列中的任务，尽快让线程池关闭
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
//传入的core的参数，唯一用到的地方，如果线程数超过理论最大容量，如果core是true跟最大核心线程数比较，否则跟最大线程数比较
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//通过CAS自旋，增加线程数+1，增加成功跳出双层循环，继续往下执行
if (compareAndIncrementWorkerCount(c))
break retry;
//检测当前线程状态如果发生了变化，则继续回到retry，重新开始循环
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
//走到这里，说明我们已经成功的将线程数+1了，但是真正的线程还没有被添加
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//添加线程，Worker是继承了AQS，实现了Runnable接口的包装类
w = new Worker(firstTask);
final Thread t = w.thread;这里的t就是worker自己， 构造方法以this传入了自己
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());
//检查线程状态，还是跟之前一样，只有当线程池处于RUNNING，或者处于SHUTDOWN并且firstTask==null的时候，这时候创建Worker来加速处理队列中的任务
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
//线程只能被start一次
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//workers是一个HashSet，添加我们新增的Worker
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//启动Worker
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
我们看一下重点代码w = new Worker(firstTask);
查看worker的构造器代码
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
这里给worker一个thread属性这个线程的具体实现是当前的worker，也就是new Thread(new Runnable(){public void run(){｝})
所以在t.start();启动worker的就是启动worker 这个线程，所以就会运行run方法，接下来查看run 方法
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 {
//如果task不是null，或者去队列中取任务，注意这里会阻塞，后面会分析getTask方法
while (task != null || (task = getTask()) != null) {
//这个lock在这里是为了如果线程被中断，那么会抛出InterruptedException，而退出循环，结束线程
w.lock();
//判断线程是否需要中断
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//任务开始执行前的hook方法
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();//这里就是直接调用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 {
////任务开始执行后的hook方法
afterExecute(task, thrown);
}
} finally {
task = null;//清空task
w.completedTasks++;//完成数添加
w.unlock();
}
}
completedAbruptly = false;
} finally {
//Worker退出
processWorkerExit(w, completedAbruptly);
}
}
while 循环 直接执行线程的run（）方法，该线程执行完第一个任务之后不断去获取新的任务，这样子就实现了worker的复用
