Это 9-й день моего участия в Gengwen Challenge.Подробности мероприятия смотрите:Обновить вызов
принцип выполнения()
Этапы выполнения следующие:
Если workerCount меньше, чем corePoolSize, задача создается как первая задача для выполнения задачи. Добавьте задачу в очередь.Если добавление прошло успешно, еще нужно проверить.Перед тем как поставить задачу,
Если пул потоков закрыт, удалите задачу из очереди;
Если в текущем пуле потоков нет рабочих потоков, а задача только что добавлена в очередь, убедитесь, что по крайней мере один рабочий поток в пуле потоков может обработать задачу.
Если задачу не удается добавить в очередь, попробуйте добавить новый поток для выполнения задачи, но это может не увенчаться успехом. Возможно, пул потоков закрыт или пул потоков достиг насыщения (maximumPoolSize). Если это не удается , выполняется политика отказа.
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
// 高3位表示状态，低29位任务数量。
//工作线程小于核心线程数，创建新的线程。
if (workerCountOf(c) < corePoolSize) {
//创建新的worker立即执行command，轮询workQueue处理task。
if (addWorker(command, true))
return;
c = ctl.get();
}
//线程池在运行状态且可以将task插入队列
//第一次校验线程池在运行状态
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
//第二次校验，防止在第一次校验通过后线程池关闭。如果线程池关闭，在队列中删除task并拒绝task
if (! isRunning(recheck) && remove(command))
reject(command);
//如果线程数=0（线程都死掉了，比如：corePoolSize=0），新建线程且未指定firstTask，仅仅去轮训workQueue
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
//线程队列已满，尝试创建新线程执行task，创建失败后拒绝task，创建失败原因：1.线程池关闭；2.线程数已经达到maxPoolSize
else if (!addWorker(command, false))
reject(command);
}
метод addWorker
Метод addWorker состоит из двух основных частей.
Первый: определить, можно ли создать работника. По спину, CAS, ctl и т. д. судят, продолжать создавать или возвращать false, а период спина обычно очень короткий.
Два: синхронно создать работника и запустить поток. addWorker определит, можно ли добавить новый рабочий поток в соответствии с текущим рабочим состоянием пула потоков и заданными границами (corePoolSize и maxPoolSize). Каждый из этих объектов Worker представляет собой очередь AQS.
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
//外层循环判断线程池的状态
for (;;) {
//在进行获取一次上下文操作机制
int c = ctl.get();
int rs = runStateOf(c);
//线程池状态
// Check if queue empty only if necessary.
// 线程池状态:RUNNING = -1、SHUTDOWN = 0、 STOP = 1、TIDYING = 2、TERMINATED = 3
//线程池至少是shutdown状态
if (rs >= SHUTDOWN &&
// 除了线程池正在关闭（shutdown），
// 队列里还有未处理的task的情况，其他都不能添加
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
//内层循环判断是否到达容量上限，worker+1
for (;;) {
int wc = workerCountOf(c);//worker数量
//worker大于Integer最大上限
//或到达边界上限
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//CAS worker+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
// CAS失败workerCount被其他线程改变，
// 重新尝试内层循环CAS对workerCount+1
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
final ReentrantLock mainLock = this.mainLock;
w = new Worker(firstTask);
//1.state置为-1，Worker继承了AbstractQueuedSynchronizer.
//2.设置firstTask属性.
//3.Worker实现了Runable接口，将this作为入参创建线程.
final Thread t = w.thread;
if (t != null) {
//addWorker需要加锁
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
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);//workers是HashSet<Worker>
//设置最大线程池大小
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;
}
Worker
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
Метод Worker реализует Runnable, который присваивает созданный поток потоку. setState(-1) на первом шаге — установить для состояния синхронизации в AQS значение -1, а пул потоков использует состояние синхронизации в AQS, чтобы определить, можно ли прервать рабочий поток. -1: значение инициализации.В это время рабочий поток не запущен, и прерывание не требуется;0: означает принятие прерывания, и рабочий поток в это время находится в состоянии ожидания;1: означает, что рабочий поток выполняет задачи в это время. Он реализует нереентерабельный мьютекс, а нереентерабельность предназначена для того, чтобы некоторые методы управления пулом потоков не получали повторные блокировки. Обратите внимание, что цель реализации блокировки Worker отличается от цели традиционных блокировок. В основном это контроль над тем, можно ли прервать поток, и другой мониторинг, например, активен ли поток (выполняется ли задача).
private final class Worker extends AbstractQueuedSynchronizer implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
//每个worker有自己的内部线程，ThreadFactory创建失败时是null
//封装任务线程机制
final Thread thread;
//初始化任务，可能是null
Runnable firstTask;
//每个worker的完成任务数
volatile long completedTasks;
Worker(Runnable firstTask) {
// inhibit interrupts until runWorker //状态置为-1，如果中断线程需要CAS将state 从0-
>1，以此来保证能只中断从workerQueue getTask的线程
setState(-1); // 禁止线程在启动前被打断
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
//核心方法
public void run() {
//首先执行w.unlock,就是把state置为0，对该线程的中断就可以进行了
runWorker(this);
}
// state = 0 代表未锁；state = 1 代表已锁
protected boolean isHeldExclusively() {
return getState() != 0;
}
// 在setCorePoolSize/shutdown等方法中断worker线程时需要调用该方法，
// 确保中断的是从workerQueue getTask的线程
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); }
//调用tryRelease修改state=0，LockSupport.unpark(thread) 下一个等待锁的线程
public boolean isLocked() { return isHeldExclusively(); }
// interrupt已启动线程
void interruptIfStarted() {
Thread t;
// 初始化是 state = -1，不会被interrupt
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
Метод run() Worker фактически вызывает runWorker(this), запускает рабочий поток (обратите внимание, что это не запускаемая напрямую задача) и выполняет задачу в рабочем потоке. Здесь зацикливаются рабочие потоки, многократно извлекающие задачи из очереди и выполняющие их.
Перед запуском всех задач необходимо получить блокировки, чтобы гарантировать, что задачи не будут прерваны во время выполнения, если только пул потоков не остановлен (Stop). Обобщенно следующим образом: 1. Класс Worker в основном отвечает за управление состоянием запущенного потока. 2. Worker наследует AQS и реализует простые операции получения и снятия блокировки. чтобы избежать прерывания строк, ожидающих выполнения задач Когда процесс запущен, прервите работающий поток (поток только что запущен и еще не начал выполнение задачи). 3. Реализация нереентерабельных блокировок самостоятельно, чтобы избежать прерывания работающего потока при реализации метода управления состоянием управления пулом потоков, например: setCorePoolSize. setCorePoolSize может вызвать interruptIdleWorkers(), который вызовет метод tryLock() рабочего потока, чтобы прервать поток Реализация блокировки самостоятельно может гарантировать, что рабочий поток не будет прерван до его запуска.
Основная задача runWorker — обработать задачи в цикле, а если задачи нет — перейти к getTask(), код показывает, как показано ниже:
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 允许被 interrupt
boolean completedAbruptly = true;
try {
//循环获取任务直至 task = null （线程池关闭、超时等）
// 注意这里的getTask()方法，我们配置的阻塞队列会在这里起作用
while (task != null || (task = getTask()) != null) {
w.lock(); // 执行任务前上锁
// 如果线程池停止(处于STOP状态或者TIDYING、TERMINATED状态时)，设置当前线程处于中断状态如果不是(线程就处于RUNNING或者SHUTDOWN状态),确保线程不中断,重新检查当前线程池的状态是否大于等于STOP状态
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//在线程运行前调用
beforeExecute(wt, task);
Throwable thrown = null;
try {
//这里执行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()
Получение Задачи из очереди условно разделено на следующие шаги.
Оцените состояние пула потоков и очереди. Если состояние пула потоков выше STOP, пул потоков не обрабатывает задачи в очереди в это время или пул потоков находится в состоянии SHUTDOWN, но очередь пуста (SHUTDOWN no дольше принимает новые задачи), workerCount уменьшается на 1, возвращает null, обратите внимание, что переменная только уменьшается на 1 в это время, на самом деле рабочий поток не завершает реальное завершение в processWorkerExit(w, completeAbruptly) ;. Если проверка состояния пройдена, решается, следует ли выполнять перезапуск потока.Если количество workerCount необходимо уменьшить на 1, после успеха возвращается null. В соответствии с заданным временем (время указывает на то, что необходимо выполнить восстановление простаивающего потока по истечении времени ожидания), выберите, следует ли ждать в течение ограниченного времени или блокировать получение задач из очереди.
Метод работает по настроенной workQueue, а его принцип блокировки и принцип тайм-аута реализованы на основе очереди блокировки и подробно описываться не будут.
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);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
//（当前线程数是否大于最大线程数或者）
//且（线程数大于1或者任务队列为空）
//这里有个问题(timed && timedOut)timedOut = false，好像(timed && timedOut)一直都是false吧
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;
}
}
}
Суммировать
Ядром ThreadPoolExecutor является Worker, который реализует AbstractQueuedSynchronizer и наследует Runnable. Использование блокировки aqs очень умно.