1.场景
线程池运用DiscardOldestPolicy回绝战略,堵塞行列运用ArrayBlockingQueue,发现在某些景象下对于得到的Future,调用get()办法当时线程会一向堵塞。
为了便于了解,将实践情景笼统为下面的代码:
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
1,
1,
1,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(1),
Executors.defaultThreadFactory(),
new ThreadPoolExecutor.DiscardOldestPolicy());//新建线程池时中心线程数及最大线程数都设置为1,堵塞行列运用ArrayBlockingQueue,回绝战略为DiscardOldestPolicy
public void doBusiness(){
Task task1 = new Task();
Task task2 = new Task();
Task task3 = new Task();
Future<Boolean> future1 = threadPoolExecutor.submit(task1);//当时作业线程为0,会新建一个worker作为作业线程,并履行task1
Future<Boolean> future2 = threadPoolExecutor.submit(task2);//当时中心线程数已满,会将使命放入堵塞行列
Future<Boolean> future3 = threadPoolExecutor.submit(task3);
/*当时中心线程已满而且堵塞行列已满,execute()时会调用ThreadPoolExecutord的addWorker(command,false),由
于目前task1还没履行完,则作业线程数量为1,现已达到了最大线程数,则addWorker(command,false)返回false,
触发对应的回绝战略,会从堵塞行列中移除task2对应的使命(堵塞行列中并不是直接放的task2,而是以task2为入
参结构的一个FutureTask,拜见AbstarctExecutorService的submit(Callable<T> task)办法*/
try{
boolean result = future2.get();
System.out.println(result);
} catch (ExecutionException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Test
public void test_doBusiness(){
doBusiness();//进口
}
private class Task implements Callable<Boolean>{
@Override
public Boolean call() throws Exception {
try {
Thread.sleep(1000);//模拟事务履行
return true;
}catch(Exception e){
e.printStackTrace();
}
return true;
}
}
2. 原因剖析
经过上面代码咱们明白了堵塞行列会将task2对应的使命移除,那么为何移除之后调用get()办法线程会一向堵塞呢?
其实Future future2= threadPoolExecutor.submit(task2)实践会调用AbstractExecutorService的submit(Callable task)办法,而且终究返回的future2实践是一个FutureTask类型。
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
因此,咱们直接看FutureTask的get()办法
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
因为future2现已从堵塞行列中移除,而且从始至终都没有作业线程履行它,即FutureTask的状况一向都为NEW状况,其会进入awaitDone(false,0L)中,接下列咱们追踪该办法。
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)//第一次进for循环时q==null,进入到该分支
q = new WaitNode();
else if (!queued)//第二次进for循环时queue为false,则运用CAS将q置为waiters的头结点
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else//将q置为头结点后,终究会进入这里调用park()办法,堵塞当时线程
LockSupport.park(this);
}
从上面的代码可以看出调用future2.get()后会一向堵塞在park()办法处,这便是本次问题呈现的原因,
3.总结
本次问题呈现主要是同时满意了以下几点:
1)运用了有界的堵塞行列ArrayBlockingQueue
2)作业线程达到了线程池装备的最大线程数
3)回绝战略运用了DiscardOldestPolicy(运用DiscardPolicy也会呈现这个问题)
4.考虑
咱们日常运用线程池提交使命后,假如在使命履行完结之前调用future的get()办法,当时线程会进入堵塞状况,当使命履行完结后,才会将当时线程唤醒,如何从代码上剖析该流程?
首先当使命提交到线程池,假如使命当时在堵塞行列中,则FutureTask的状况仍然像上面的状况一样,是处于New状况,调用get()办法仍然会到达LockSupport.park(this)处,将当时线程堵塞。什么时候才会将当时线程唤醒了?那就是当存在作业线程Worker目前分配的使命履行完结后,其会去调用Worker类的getTask()办法从堵塞行列中拿到该使命,并履行该使命的run()办法,下面是FutureTask的run()办法
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);//假如使命履行成功,则调用set(V result)办法
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
其会在履行成功后,调用set(V result)办法
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();//
}
}
然后将FutureTask状况置为NORMAL(FutureTask的状况要和ThreadPoolExecutor的状况区分开),接着调用finishCompletion()办法
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
for (;;) {
Thread t = q.thread;//q在await()办法中设置的,其值为调用get()办法的线程
if (t != null) {
q.thread = null;
LockSupport.unpark(t);//唤醒该线程
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();//熟悉的钩子办法
callable = null; // to reduce footprint
}
在finishCompletion中唤起因get()而堵塞的线程。
以上