關(guān)于PriorityBlockingQueue中隊(duì)列操作

編者按：筆者在使用PriorityBlockingQueue實(shí)現(xiàn)按照優(yōu)先級(jí)處理任務(wù)時(shí)遇到一類(lèi)NPE問(wèn)題，經(jīng)過(guò)分析發(fā)現(xiàn)根本原因是在任務(wù)出隊(duì)列時(shí)調(diào)用比較器異常，進(jìn)而導(dǎo)致后續(xù)任務(wù)出隊(duì)列拋出NullPointerException。本文通過(guò)完整的案例復(fù)現(xiàn)來(lái)演示在什么情況會(huì)觸發(fā)該問(wèn)題，同時(shí)給出了處理建議。希望讀者在編程時(shí)加以借鑒，避免再次遇到此類(lèi)問(wèn)題。

背景知識(shí)

PriorityBlockingQueue是一個(gè)無(wú)界的基于數(shù)組的優(yōu)先級(jí)阻塞隊(duì)列，使用一個(gè)全局ReentrantLock來(lái)控制某一時(shí)刻只有一個(gè)線程可以進(jìn)行元素出隊(duì)和入隊(duì)操作，并且每次出隊(duì)都返回優(yōu)先級(jí)別最高的或者最低的元素。PriorityBlockingQueue通過(guò)以下兩種方式實(shí)現(xiàn)元素優(yōu)先級(jí)排序：

1. 入隊(duì)元素實(shí)現(xiàn)Comparable接口來(lái)比較元素優(yōu)先級(jí)；
2. PriorityBlockingQueue構(gòu)造函數(shù)指定Comparator來(lái)比較元素優(yōu)先級(jí)；

關(guān)于PriorityBlockingQueue中隊(duì)列操作的部分，基本和PriorityQueue邏輯一致，只不過(guò)在操作時(shí)加鎖了。在本文中我們主要關(guān)注PriorityBlockingQueue出隊(duì)的take方法，該方法通過(guò)調(diào)用dequeue方法將元素出隊(duì)列。當(dāng)沒(méi)有元素可以出隊(duì)的時(shí)候，線程就會(huì)阻塞等待。

publicEtake()throwsInterruptedException{
finalReentrantLocklock=this.lock;
lock.lockInterruptibly();
Eresult;
try{
//嘗試獲取最小元素，即小頂堆第一個(gè)元素，然后重新排序，如果不存在表示隊(duì)列暫無(wú)元素，進(jìn)行阻塞等待。
while((result=dequeue())==null)
notEmpty.await();
}finally{
lock.unlock();
}
returnresult;
}

現(xiàn)象

在某個(gè)業(yè)務(wù)服務(wù)中使用PriorityBlockingQueue實(shí)現(xiàn)按照優(yōu)先級(jí)處理任務(wù)，某一天環(huán)境中的服務(wù)突然間不處理任務(wù)了，查看后臺(tái)日志，發(fā)現(xiàn)一直拋出NullPointerException。將進(jìn)程堆dump出來(lái)，使用MAT發(fā)現(xiàn)某個(gè)PriorityBlockingQueue中的size值比實(shí)際元素個(gè)數(shù)多1個(gè)（入隊(duì)時(shí)已經(jīng)對(duì)任務(wù)進(jìn)行非空校驗(yàn)）。

異常堆棧如下：

java.lang.NullPointerException
atjava.util.concurrent.PriorityBlockingQueue.siftDownComparable(PriorityBlockingQueue.java:404)
atjava.util.concurrent.PriorityBlockingQueue.dequeue(PriorityBlockingQueue.java:333)
atjava.util.concurrent.PriorityBlockingQueue.take(PriorityBlockingQueue.java:548)
...

MAT結(jié)果：

原因分析

在此我們分析下PriorityBlockingQueue是如何出隊(duì)列的，PriorityBlockingQueue最終通過(guò)調(diào)用dequeue方法出隊(duì)列，dequeue方法處理邏輯如下：

1. 將根節(jié)點(diǎn)（array[0]）賦值給result；
2. array[n] 賦值給 arrary[0]；
3. 將 array[n] 設(shè)置為 null；
4. 調(diào)用siftDownComparable或siftDownUsingComparator對(duì)隊(duì)列元素重新排序；
5. size大小減1；
6. 返回result；

如果在第4步中出現(xiàn)異常，就會(huì)出現(xiàn)隊(duì)列中的元素個(gè)數(shù)比實(shí)際的元素個(gè)數(shù)多1個(gè)的現(xiàn)象。此時(shí)size未發(fā)生改變，arry[n]已經(jīng)被置為null，再進(jìn)行siftDown操作時(shí)就會(huì)拋出NullPointerException。繼續(xù)分析第4步中在什么情況下會(huì)出現(xiàn)異常，通過(guò)代碼走讀我們可以發(fā)現(xiàn)只有在調(diào)用Comparable#compareTo或者Comparator#compare方法進(jìn)行元素比較的時(shí)候才可能出現(xiàn)異常。這塊代碼的處理邏輯和業(yè)務(wù)相關(guān)，如果業(yè)務(wù)代碼處理不當(dāng)拋出異常，就會(huì)導(dǎo)致上述現(xiàn)象。

/**
*Mechanicsforpoll().Callonlywhileholdinglock.
*/
privateEdequeue(){
intn=size-1;
if(n0)
returnnull;
else{
Object[]array=queue;
Eresult=(E)array[0];//step1
Ex=(E)array[n];//step2
array[n]=null;//step3
ComparatorsuperE>cmp=comparator;
if(cmp==null)//step4 如果指定了comparator，就按照指定的comparator來(lái)比較。否則就按照默認(rèn)的
siftDownComparable(0,x,array,n);
else
siftDownUsingComparator(0,x,array,n,cmp);
size=n;//step5
returnresult;//step6
}
}

privatestaticvoidsiftDownComparable(intk,Tx,Object[]array,intn){
if(n>0){
ComparablesuperT>key=(ComparablesuperT>)x;
inthalf=n>>>1;
while(kintchild=(k<1)+1;
Objectc=array[child];
intright=child+1;
if(rightsuperT>)c).compareTo((T)array[right])>0)
c=array[child=right];
if(key.compareTo((T)c)<=?0)
break;
array[k]=c;
k=child;
}
array[k]=key;
}
}
privatestaticvoidsiftDownUsingComparator(intk,Tx,Object[]array,intn,
ComparatorsuperT>cmp){
if(n>0){
inthalf=n>>>1;
while(kintchild=(k<1)+1;
Objectc=array[child];
intright=child+1;
if(right0)
c=array[child=right];
if(cmp.compare(x,(T)c)<=?0)
break;
array[k]=c;
k=child;
}
array[k]=x;
}
}

復(fù)現(xiàn)代碼

importjava.util.ArrayList;
importjava.util.List;
importjava.util.concurrent.PriorityBlockingQueue;

publicclassPriorityBlockingQueueTest{
staticclassEntityimplementsComparable<Entity>{
privateintid;
privateStringname;
privatebooleanflag;

publicvoidsetFlag(booleanflag){
this.flag=flag;
}

publicEntity(intid,Stringname){
this.id=id;
this.name=name;
}

@Override
publicintcompareTo(Entityentity){
if(flag){
thrownewRuntimeException("TestException");
}
if(entity==null||this.id>entity.id){
return1;
}
returnthis.id==entity.id?0:-1;
}
}

publicstaticvoidmain(String[]args){
intnum=5;
PriorityBlockingQueuepriorityBlockingQueue=newPriorityBlockingQueue<>();
Listentities=newArrayList<>();
for(inti=0;inewEntity(i,"entity"+i);
entities.add(entity);
priorityBlockingQueue.offer(entity);
}

entities.get(num-1).setFlag(true);
intsize=entities.size();
for(inti=0;itry{
priorityBlockingQueue.take();
}catch(Exceptione){
e.printStackTrace();
}
}
}

執(zhí)行結(jié)果如下：

java.lang.RuntimeException:TestException
atPriorityBlockingQueueTest$Entity.compareTo(PriorityBlockingQueueTest.java:31)
atPriorityBlockingQueueTest$Entity.compareTo(PriorityBlockingQueueTest.java:8)
atjava.util.concurrent.PriorityBlockingQueue.siftDownComparable(PriorityBlockingQueue.java:404)
atjava.util.concurrent.PriorityBlockingQueue.dequeue(PriorityBlockingQueue.java:333)
atjava.util.concurrent.PriorityBlockingQueue.take(PriorityBlockingQueue.java:548)
atPriorityBlockingQueueTest.main(PriorityBlockingQueueTest.java:71)
java.lang.NullPointerException
atjava.util.concurrent.PriorityBlockingQueue.siftDownComparable(PriorityBlockingQueue.java:404)
atjava.util.concurrent.PriorityBlockingQueue.dequeue(PriorityBlockingQueue.java:333)
atjava.util.concurrent.PriorityBlockingQueue.take(PriorityBlockingQueue.java:548)
atPriorityBlockingQueueTest.main(PriorityBlockingQueueTest.java:71)

規(guī)避方案

可以通過(guò)以下兩種方法規(guī)避：

• 在take方法出現(xiàn)NPE時(shí)，清除隊(duì)列元素，將未處理的元素重新進(jìn)入隊(duì)列；
• 在 Comparable#compareTo 或 Comparator#compare 方法中做好異常處理，對(duì)異常情況進(jìn)行默認(rèn)操作；

建議使用后者。

案例引申

使用PriorityBlockingQueue作為緩存隊(duì)列來(lái)創(chuàng)建線程池時(shí)，使用submit提交任務(wù)會(huì)出現(xiàn) java.lang.ClassCastException: java.util.concurrent.FutureTask cannot be cast to 異常，而使用execute沒(méi)有問(wèn)題。

觀察submit源碼可以發(fā)現(xiàn)在submit內(nèi)部代碼會(huì)將Runable封裝成RunnableFuture對(duì)象，然后調(diào)用execute提交任務(wù)。

publicFuturesubmit(Runnabletask){
if(task==null)thrownewNullPointerException();
RunnableFutureftask=newTaskFor(task,null);
execute(ftask);
returnftask;
}

以Comparable為例，任務(wù)入隊(duì)列時(shí)，最終會(huì)調(diào)用siftUpComparable方法。該方法第一步將RunnableFuture強(qiáng)轉(zhuǎn)為Comparable類(lèi)型，而RunnableFuture類(lèi)未實(shí)現(xiàn)Comparable接口，進(jìn)而拋出ClassCastException異常。

publicbooleanoffer(Ee){
if(e==null)
thrownewNullPointerException();
finalReentrantLocklock=this.lock;
lock.lock();
intn,cap;
Object[]array;
while((n=size)>=(cap=(array=queue).length))
tryGrow(array,cap);
try{
ComparatorsuperE>cmp=comparator;
if(cmp==null)
siftUpComparable(n,e,array);
else
siftUpUsingComparator(n,e,array,cmp);
size=n+1;
notEmpty.signal();
}finally{
lock.unlock();
}
returntrue;
}

privatestaticvoidsiftUpComparable(intk,Tx,Object[]array){
ComparablesuperT>key=(ComparablesuperT>)x;
while(k>0){
intparent=(k-1)>>>1;
Objecte=array[parent];
if(key.compareTo((T)e)>=0)
break;
array[k]=e;
k=parent;
}
array[k]=key;
}

這也是常見(jiàn)的比較器調(diào)用異常案例，本文不再贅述，可自行參考其他文章。

總結(jié)

在使用PriorityBlockingQueue時(shí)，注意在比較器中做好異常處理，避免出現(xiàn)類(lèi)似問(wèn)題。

審核編輯 ：李倩