Java中ThreadPoolExecutor拒絕策略踩坑

更新時(shí)間：2023年06月15日 10:20:13 作者：星辰之行

本文主要介紹了Java中ThreadPoolExecutor拒絕策略踩坑，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

1.場(chǎng)景

線程池使用DiscardOldestPolicy拒絕策略，阻塞隊(duì)列使用ArrayBlockingQueue,發(fā)現(xiàn)在某些情形下對(duì)于得到的Future，調(diào)用get()方法當(dāng)前線程會(huì)一直阻塞。
為了便于理解，將實(shí)際情景抽象為下面的代碼：

ThreadPoolExecutor threadPoolExecutor ?= new ThreadPoolExecutor(
? ? ? ? 1,
? ? ? ? 1,
? ? ? ? 1,
? ? ? ? TimeUnit.SECONDS,
? ? ? ? new ArrayBlockingQueue<>(1),
? ? ? ? Executors.defaultThreadFactory(),
? ? ? ? new ThreadPoolExecutor.DiscardOldestPolicy());//新建線程池時(shí)核心線程數(shù)及最大線程數(shù)都設(shè)置為1，阻塞隊(duì)列使用ArrayBlockingQueue，拒絕策略為DiscardOldestPolicy
public void doBusiness(){
? ? Task task1 = new Task();
? ? Task task2 = new Task();
? ? Task task3 = new Task();
? ? Future<Boolean> future1 = threadPoolExecutor.submit(task1);//當(dāng)前工作線程為0，會(huì)新建一個(gè)worker作為工作線程，并執(zhí)行task1
? ? Future<Boolean> future2 = threadPoolExecutor.submit(task2);//當(dāng)前核心線程數(shù)已滿，會(huì)將任務(wù)放入阻塞隊(duì)列
? ? Future<Boolean> future3 = threadPoolExecutor.submit(task3);
? ? /*當(dāng)前核心線程已滿并且阻塞隊(duì)列已滿，execute()時(shí)會(huì)調(diào)用ThreadPoolExecutord的addWorker(command,false)，由
? ? 于目前task1還沒執(zhí)行完，則工作線程數(shù)量為1，已經(jīng)達(dá)到了最大線程數(shù)，則addWorker(command,false)返回false，
? ? 觸發(fā)對(duì)應(yīng)的拒絕策略，會(huì)從阻塞隊(duì)列中移除task2對(duì)應(yīng)的任務(wù)(阻塞隊(duì)列中并不是直接放的task2，而是以task2為入
? ? 參構(gòu)造的一個(gè)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);//模擬業(yè)務(wù)執(zhí)行
? ? ? ? ? ? return true;
? ? ? ? }catch(Exception e){
? ? ? ? ? ? e.printStackTrace();
? ? ? ? }
? ? ? ? return true;
? ? }
}

2. 原因分析

通過上面代碼我們明白了阻塞隊(duì)列會(huì)將task2對(duì)應(yīng)的任務(wù)移除，那么為何移除之后調(diào)用get()方法線程會(huì)一直阻塞呢？
其實(shí)Future future2= threadPoolExecutor.submit(task2)實(shí)際會(huì)調(diào)用AbstractExecutorService的submit(Callable task)方法，并且最終返回的future2實(shí)際是一個(gè)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已經(jīng)從阻塞隊(duì)列中移除，并且從始至終都沒有工作線程執(zhí)行它，即FutureTask的狀態(tài)一直都為NEW狀態(tài)，其會(huì)進(jìn)入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)//第一次進(jìn)for循環(huán)時(shí)q==null，進(jìn)入到該分支
? ? ? ? ? ? q = new WaitNode();
? ? ? ? else if (!queued)//第二次進(jìn)for循環(huán)時(shí)queue為false，則使用CAS將q置為waiters的頭結(jié)點(diǎn)
? ? ? ? ? ? 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置為頭結(jié)點(diǎn)后，最終會(huì)進(jìn)入這里調(diào)用park()方法，阻塞當(dāng)前線程
? ? ? ? ? ? LockSupport.park(this);
? ? }

從上面的代碼可以看出調(diào)用future2.get()后會(huì)一直阻塞在park()方法處，這便是本次問題出現(xiàn)的原因，

3.總結(jié)

本次問題出現(xiàn)主要是同時(shí)滿足了以下幾點(diǎn)：

1)使用了有界的阻塞隊(duì)列ArrayBlockingQueue
2)工作線程達(dá)到了線程池配置的最大線程數(shù)
3)拒絕策略使用了DiscardOldestPolicy(使用DiscardPolicy也會(huì)出現(xiàn)這個(gè)問題)

4.思考

我們?nèi)粘Ｊ褂镁€程池提交任務(wù)后，如果在任務(wù)執(zhí)行完成之前調(diào)用future的get()方法，當(dāng)前線程會(huì)進(jìn)入阻塞狀態(tài)，當(dāng)任務(wù)執(zhí)行完成后，才會(huì)將當(dāng)前線程喚醒，如何從代碼上分析該流程？

首先當(dāng)任務(wù)提交到線程池，如果任務(wù)當(dāng)前在阻塞隊(duì)列中，則FutureTask的狀態(tài)依然像上面的情況一樣，是處于New狀態(tài)，調(diào)用get()方法依然會(huì)到達(dá)LockSupport.park(this)處，將當(dāng)前線程阻塞。什么時(shí)候才會(huì)將當(dāng)前線程喚醒了？那就是當(dāng)存在工作線程Worker目前分配的任務(wù)執(zhí)行完成后，其會(huì)去調(diào)用Worker類的getTask()方法從阻塞隊(duì)列中拿到該任務(wù)，并執(zhí)行該任務(wù)的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);//如果任務(wù)執(zhí)行成功，則調(diào)用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);
? ? }
}

其會(huì)在執(zhí)行成功后，調(diào)用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狀態(tài)置為NORMAL(FutureTask的狀態(tài)要和ThreadPoolExecutor的狀態(tài)區(qū)分開),接著調(diào)用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()方法中設(shè)置的，其值為調(diào)用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()而阻塞的線程。

到此這篇關(guān)于Java中ThreadPoolExecutor拒絕策略踩坑的文章就介紹到這了,更多相關(guān)Java ThreadPoolExecutor拒絕策略內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: