线程池在日常开发中多多少少都会接触和使用.
其中和线程池关系最为紧密的一个就是阻塞队列,用于存储提交到线程池中的任务.
关于向阻塞队列中添加任务和获取任务会涉及到很多方法,如下
image.png
那么当我们向线程池提交任务的时候,它会调用上面的哪个方法呢?
// 代码位置: java.util.concurrent.ThreadPoolExecutor#execute
public void execute(Runnable command) {
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// 向线程池队列中添加任务
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(command)方法.
也就是说,它并不会阻塞提交任务的线程.
在线程池中的线程会不停的从阻塞队列中获取任务,那么它们又是调用的哪个方法呢?
// 代码位置: java.util.concurrent.ThreadPoolExecutor#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;
}
}
}
从源码中我们知道,线程池中的线程在向阻塞队列获取任务时,通过超时获取或者阻塞获取的方式.
image.png
讨论第二个注意点
我们在学习ReentrantLock的时候,手动加锁和释放锁必须是成对出现的,这也是我们大家一贯的认知.
// 代码位置: java.util.concurrent.ThreadPoolExecutor#runWorker
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 ((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);
}
}
代码中,我们可以看到lock()和unlokc()成对出现了.可是奇怪的是,在方法开始处还有一个unlock()调用.
w.unlock(); // allow interrupts
在没有出现lock()的情况下,居然出现了unlock()调用.作者还注释了allow interrupts.
是不是觉得有点和之前自己的认知有点反差.
这里我们简单介绍下线程池中的Worker这个类.
// 代码位置: java.util.concurrent.ThreadPoolExecutor.Worker
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable {
}
它继承了AQS,自己实现了加锁和释放锁,并没有使用ReentrantLock.
它并没有像ReentrantLock那样是可重入的,它不允许重入.
而且在它的构造函数中,直接将state设置成-1,并没有使用默认值0.
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
在构造函数中,将state默认值本应该为0,被设置成了-1,相当于lock()的语义.
而且源码注释上也说明了,禁止中断直到运行runWorker. 因此我们才会看到上面那个一开始就调用了unlock()方法.
关于基于AQS实现各种锁,以及操作state这个值,我们必须灵活使用.不能被常规认识所蒙蔽.
网友评论