Haiqingd 还在输出

问题背景

问题背景部分内容比较长，和 BlockingPriorityQueue 无关，不感兴趣的读者可以跳过，内容是关于如何定位到问题出在 BlockingPriorityQueue 上的。

公司内一个项目有通过 netty 打开监听端口，接收到报文，经过一些列处理后，会响应写回对应的通道channel。但是在预投产环境做验证时，发现端口明明是通的，发送报文却迟迟无法获取到响应。

问题定位

关于线程池的怀疑

首先想到的是线程池有问题，对于 netty 接收到的报文，我们的系统会将其放进一个报文队列中，然后有一个固定线程一直做轮询，将报文队列逐个取出做处理。我们恰好最近改动到过这个线程池，怀疑是这个轮询线程挂掉了（实际我们这个端口接收的是心跳报文，因为打印的日志级别过低未在预投产环境打印出，从而无法判定到底是没有接收到，还是没有轮询）。

• 怀疑的排除：

jstack <pid> > 1.jstack

通过 jstack 获取到对应的线程信息。由于这个线程是在做死循环轮询的，所以只要它没有挂，就一定能够在线程信息里找到它，结果如图

我们成功发现了当前线程还活着，正在做每次轮询结束后的睡觉。从而排除了这个可能性

关于新做的改动的排查

刚刚已经拿到了线程栈信息，所以我们也顺便看了看新改动点相关的 jstack，同样的，它也是使用了这个线程池，也是一个死循环的轮询工作。

但是很不幸的是，我们并没有从线程栈中获取到它应该在做的事的线程信息

• 新的线程挂了？

自然出现了如此一个怀疑思路，我们尝试本地启动程序，重新获取 jstack，出现了如下一个恐怖的图：

• 栈溢出！

看到这个图，大概率判断会导致栈溢出了，我们的方法在进行嵌套调用，那栈溢出便是 迟早的事了，我们只在线程池调用内 catch 了 Exception。众所周知，StackOverFlowError它是一个 Error，所以没捕获到。线程也就挂了

但这和接收不到报文又有什么关系呢？

上述的排查过程已经初步可以认定，很有可能是 netty 的 nio 线程没有好好干活，没有去接收消息

• 那么，netty 的线程在做什么呢？

嗯？这个 FgPriorityBlockingQueue 有点熟悉，这不刚刚的那个栈溢出一直在循环调用的对象吗。

所以结论浮现出来了，nio的线程，都被阻塞在priorityQueue 的offer方法里里，所以发给系统监听端口的请求，一直没有被处理。

BlockingPriorityQueue 一直在阻塞问题排查

在上面的背景介绍中，我们得知了两件事

1. 有一个在FgPriorityBlockingQueue 中的线程，因为 StackOverflowError 挂掉了。
2. FgPriorityBlockingQueue 在被调用 offer 方法的过程中一直在阻塞状态，等锁。

• FgPriorityBlockingQueue是什么？

只保留了其关键代码如下所示。它是一个继承了 PriorityBlockingQueue 的类，它的 removeExpiredData 和 removeExpiredTasks 通过循环调用，让队列中的元素出队

public class FgPriorityBlockingQueue extends PriorityBlockingQueue<V> {
    // ......
    private void removeExpiredData() {
        try {
                V head = this.peek();
                if (head == null) {
                    break;
                }
            	// ......
                V data = this.poll();
            }
        } finally {
            this.removeExpiredTasks();
        }
    }

    private void removeExpiredTasks() {
        try {
            // ......
			this.removeExpiredData();
        } catch (Exception e) {
            // ......
        }
    }
}

• 那，让我们看看PriorityBlockingQueue 是什么？

1. 首先，它是一个 PriorityQueue，当元素被添加进队列时，将会调用排序方法
2. 其次，它是一个 Bloking，它通过锁来支持并发入队出队，防止扩容时的并发访问导致的线程安全问题。
3. 它是无界的，只要不超过int上限

来看看它的 offer 和 poll 方法

public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    
    /**
     * Inserts the specified element into this priority queue.
     * As the queue is unbounded, this method will never return {@code false}.
     *
     * @param e the element to add
     * @return {@code true} (as specified by {@link Queue#offer})
     * @throws ClassCastException if the specified element cannot be compared
     *         with elements currently in the priority queue according to the
     *         priority queue's ordering
     * @throws NullPointerException if the specified element is null
     */
    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;
    }
    
    
    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
    
}

假设推演

这张图里看到，在 offer 方法的 lock.lock(); 行阻塞住了，线程处于 WAITING 态

ReentrantLock 的非公平锁代码如下

从我们的用法分析，是谁获取到的锁呢？结合我们代码中的用法，提出两个假设

1. poll 之后获取的锁，没有释放。我大胆假设为：我们抛出StackOverFlow的那个线程，恰恰在lock.lock()的深层嵌套处，抛出的error。而此时未调用unlock，就抛出了异常。（可惜的是，线程没有捕获这个 Error，看不见相关日志）
2. 另一个线程的 offer 方法获取了锁，由于出队的线程已经挂掉，所以队列内的元素很多，排序花费时间久。

复现验证

1. lock.lock()中的深层调用

测试代码如下

public static void main(String[] args) {
    PriorityBlockingQueue<String> queue = new PriorityBlockingQueue<>();
    queue.offer("init");
    new Thread(() -> {
        queue.poll();
    }).start();
    
    new Thread(() -> {
        queue.offer("123");
    }).start();
}

我在两个thread中打断点，并且在ReentrantLock的setExclusiveOwnerThread方法，抛出Error。

Connected to the target VM, address: '127.0.0.1:59181', transport: 'socket'
Exception in thread "Thread-0" java.lang.Error
	at java.util.concurrent.locks.AbstractOwnableSynchronizer.setExclusiveOwnerThread(AbstractOwnableSynchronizer.java:74)
	at java.util.concurrent.locks.ReentrantLock$NonfairSync.lock(ReentrantLock.java:207)
	at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:285)
	at java.util.concurrent.PriorityBlockingQueue.poll(PriorityBlockingQueue.java:536)
	at com.fingard.gardpay.common.queues.FgPriorityBlockingQueue.lambda$main$0(FgPriorityBlockingQueue.java:229)
	at java.lang.Thread.run(Thread.java:748)

线程栈如图：

这是一个完美复现，并且再也不会有能对这个Queue入队的元素了

2 另一个offer正在执行

我们实际数量是不大的，从MAT分析工具中，对于dump文件拿到的队列内元素大概 < 2000个。排序算法也不复杂，即使按照 O(nlog(n))的排序算法计2000个元素，每次插入也就是15201次计算，远小于一般认为的程序执行上限10^7 - 10^8 之间的运算数量。基本排除