本文分享自华为云社区《Priority Blocking Queue比较器异常导致的NPE问题分析》，作者：谢照昆、王嘉伟。
编者按：笔者在使用 PriorityBlockingQueue 实现按照优先级处理任务时遇到一类 NPE 问题，经过分析发现根本原因是在任务出队列时调用比较器异常，进而导致后续任务出队列抛出 NullPointerException。本文通过完整的案例复现来演示在什么情况会触发该问题，同时给出了处理建议。希望读者在编程时加以借鉴，避免再次遇到此类问题。
背景知识
PriorityBlockingQueue 是一个无界的基于数组的优先级阻塞队列，使用一个全局 ReentrantLock 来控制某一时刻只有一个线程可以进行元素出队和入队操作，并且每次出队都返回优先级别最高的或者最低的元素。PriorityBlockingQueue 通过以下两种方式实现元素优先级排序：
入队元素实现 Comparable 接口来比较元素优先级;
PriorityBlockingQueue 构造函数指定 Comparator 来比较元素优先级;
关于 PriorityBlockingQueue 中队列操作的部分，基本和 PriorityQueue 逻辑一致，只不过在操作时加锁了。在本文中我们主要关注 PriorityBlockingQueue 出队的 take 方法，该方法通过调用 dequeue 方法将元素出队列。当没有元素可以出队的时候，线程就会阻塞等待。
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
E result;
try {
// 尝试获取最小元素，即小顶堆第一个元素，然后重新排序，如果不存在表示队列暂无元素，进行阻塞等待。
while ( (result = dequeue()) == null)
notEmpty.await();
} finally {
lock.unlock();
}
return result;
}
复制代码
现象
在某个业务服务中使用 PriorityBlockingQueue 实现按照优先级处理任务，某一天环境中的服务突然间不处理任务了，查看后台日志，发现一直抛出 NullPointerException。将进程堆 dump 出来，使用 MAT 发现某个 PriorityBlockingQueue 中的 size 值比实际元素个数多 1 个（入队时已经对任务进行非空校验）。
异常堆栈如下：
java.lang.NullPointerException
at java.util.concurrent.PriorityBlockingQueue.siftDownComparable(PriorityBlockingQueue.java:404)
at java.util.concurrent.PriorityBlockingQueue.dequeue(PriorityBlockingQueue.java:333)
at java.util.concurrent.PriorityBlockingQueue.take(PriorityBlockingQueue.java:548)
...
复制代码
MAT 结果：
原因分析
在此我们分析下 PriorityBlockingQueue 是如何出队列的，PriorityBlockingQueue 最终通过调用 dequeue 方法出队列，dequeue 方法处理逻辑如下：
将根节点（array[0]）赋值给 result；
array[n] 赋值给 arrary[0]；
将 array[n] 设置为 null；
调用 siftDownComparable 或 siftDownUsingComparator 对队列元素重新排序；
size 大小减 1；
返回 result；
如果在第 4 步中出现异常，就会出现队列中的元素个数比实际的元素个数多 1 个的现象。此时 size 未发生改变，arry[n]已经被置为 null，再进行 siftDown 操作时就会抛出 NullPointerException。继续分析第 4 步中在什么情况下会出现异常，通过代码走读我们可以发现只有在调用 Comparable#compareTo 或者 Comparator#compare 方法进行元素比较的时候才可能出现异常。这块代码的处理逻辑和业务相关，如果业务代码处理不当抛出异常，就会导致上述现象。
/**
* Mechanics for poll(). Call only while holding lock.
*/
private E dequeue() {
int n = size - 1;
if (n < 0)
return null;
else {
Object[] array = queue;
E result = (E) array[0]; //step1
E x = (E) array[n]; //step2
array[n] = null; //step3
Comparator<? super E> cmp = comparator;
if (cmp == null) //step4 如果指定了comparator，就按照指定的comparator来比较。否则就按照默认的
siftDownComparable(0, x, array, n);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n; //step5
return result; //step6
}
}
private static <T> void siftDownComparable(int k, T x, Object[] array, int n) {
if (n > 0) {
Comparable<? super T> key = (Comparable<? super T>)x;
int half = n >>> 1;
while (k < half) {
int child = (k << 1) + 1;
Object c = array[child];
int right = child + 1;
if (right < n && ((Comparable<? super T>) 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;
}
}
private static <T> void siftDownUsingComparator(int k, T x, Object[] array, int n,
Comparator<? super T> cmp) {
if (n > 0) {
int half = n >>> 1;
while (k < half) {
int child = (k << 1) + 1;
Object c = array[child];
int right = child + 1;
if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
c = array[child = right];
if (cmp.compare(x, (T) c) <= 0)
break;
array[k] = c;
k = child;
}
array[k] = x;
}
}
复制代码
复现代码
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue;
public class PriorityBlockingQueueTest {
static class Entity implements Comparable<Entity> {
private int id;
private String name;
private boolean flag;
public void setFlag(boolean flag) {
this.flag = flag;
}
public Entity(int id, String name) {
this.id = id;
this.name = name;
}
@Override
public int compareTo(Entity entity) {
if(flag) {
throw new RuntimeException("Test Exception");
}
if (entity == null || this.id > entity.id) {
return 1;
}
return this.id == entity.id ? 0 : -1;
}
}
public static void main(String[] args) {
int num = 5;
PriorityBlockingQueue<Entity> priorityBlockingQueue = new PriorityBlockingQueue<>();
List<Entity> entities = new ArrayList<>();
for (int i = 0; i < num; i++) {
Entity entity = new Entity(i, "entity" + i);
entities.add(entity);
priorityBlockingQueue.offer(entity);
}
entities.get(num - 1).setFlag(true);
int size = entities.size();
for (int i = 0; i < size; i++) {
try {
priorityBlockingQueue.take();
} catch (Exception e) {
e.printStackTrace();
}
}
}
复制代码
执行结果如下：
java.lang.RuntimeException: Test Exception
at PriorityBlockingQueueTest$Entity.compareTo(PriorityBlockingQueueTest.java:31)
at PriorityBlockingQueueTest$Entity.compareTo(PriorityBlockingQueueTest.java:8)
at java.util.concurrent.PriorityBlockingQueue.siftDownComparable(PriorityBlockingQueue.java:404)
at java.util.concurrent.PriorityBlockingQueue.dequeue(PriorityBlockingQueue.java:333)
at java.util.concurrent.PriorityBlockingQueue.take(PriorityBlockingQueue.java:548)
at PriorityBlockingQueueTest.main(PriorityBlockingQueueTest.java:71)
java.lang.NullPointerException
at java.util.concurrent.PriorityBlockingQueue.siftDownComparable(PriorityBlockingQueue.java:404)
at java.util.concurrent.PriorityBlockingQueue.dequeue(PriorityBlockingQueue.java:333)
at java.util.concurrent.PriorityBlockingQueue.take(PriorityBlockingQueue.java:548)
at PriorityBlockingQueueTest.main(PriorityBlockingQueueTest.java:71)
复制代码
规避方案
可以通过以下两种方法规避：
建议使用后者。
案例引申
使用 PriorityBlockingQueue 作为缓存队列来创建线程池时，使用 submit 提交任务会出现 java.lang.ClassCastException: java.util.concurrent.FutureTask cannot be cast to 异常，而使用 execute 没有问题。
观察 submit 源码可以发现在 submit 内部代码会将 Runable 封装成 RunnableFuture 对象，然后调用 execute 提交任务。
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
复制代码
以 Comparable 为例，任务入队列时，最终会调用 siftUpComparable 方法。该方法第一步将 RunnableFuture 强转为 Comparable 类型，而 RunnableFuture 类未实现 Comparable 接口，进而抛出 ClassCastException 异常。
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
int n, cap;
Object[] array;
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftUpComparable(n, e, array);
else
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}
private static <T> void siftUpComparable(int k, T x, Object[] array) {
Comparable<? super T> key = (Comparable<? super T>) x;
while (k > 0) {
int parent = (k - 1) >>> 1;
Object e = array[parent];
if (key.compareTo((T) e) >= 0)
break;
array[k] = e;
k = parent;
}
array[k] = key;
}
复制代码
这也是常见的比较器调用异常案例，本文不再赘述，可自行参考其他文章。
总结
在使用 PriorityBlockingQueue 时，注意在比较器中做好异常处理，避免出现类似问题。
后记
如果遇到相关技术问题（包括不限于毕昇 JDK），可以进入毕昇 JDK 社区查找相关资源（点击阅读原文进入官网），包括二进制下载、代码仓库、使用教学、安装、学习资料等。
点击关注，第一时间了解华为云新鲜技术~
评论