队列阻塞_Java并发|阻塞队列ArrayBlockingQueue解析

之前的文章我们学了 ConcurrentHashMap、 ConcurrentLinkedQueue 等线程安全容器，而且也说了 Java并发包中的 Concurent 开头的并发容器都是非阻塞的，是使用 CAS 自旋操作实现的线程安全。今天我们来学习实现线程安全的另一种方法：就是「阻塞」形式，即使用锁，这样的容器也被称为阻塞队列。

什么是阻塞队列

阻塞队列支持阻塞的插入和移除。

支持阻塞的插入：就是当队列满了的情况下，队列会阻塞插入元素的线程，直到队列不满。
支持阻塞的移除：就是当队列为空的情况下，队列会阻塞获取元素的线程，直到队列非空。

使用阻塞队列，我们要知道，当阻塞队列不可用时，插入和移除操作的4种处理方式，这也是我们使用的前提，你就会知道在开发中到底使用哪个方法是符合你的预期的。

方法/处理方式	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除方法	remove()	poll()	take()	poll(time,unit)
检查方法	element()	peek()	不可用	不可用

抛出异常：当队列满时，如果再往队列里插入元素，会抛出IllegalStateException("Queue full")异常。当队列空时，从队列里获取元素会抛出NoSuchElementException异常
返回特殊值：当往队列插入元素时，会返回元素是否插入成功，成功返回true。如果是移除方法，则是从队列里取出一个元素，如果没有则返回null。
一直阻塞：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到队列可用或者响应中断退出。当队列空时，如果消费者线程从队列里take元素，队列会阻塞住消费者线程，直到队列不为空。
超时退出：当阻塞队列满时，如果生产者线程往队列里插入元素，队列会阻塞生产者线程一段时间，如果超过了指定的时间，生产者线程就会退出。

「需要注意的是：」

BlockingQueue 不接受 null 值的插入，相应的方法在碰到 null 的插入时会抛出NullPointerException 异常。null 值在这里通常用于作为特殊值返回(表格中的第三列)，代表 poll 失败。所以，如果允许插入 null 值的话，那获取的时候，就不能很好地用 null 来判断到底是代表失败，还是获取的值就是 null 值。

ArrayBlockingQueue 概述

❝

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出(FIFO)的原则对元素进行排序。

❞

ArrayBlockingQueue 默认情况下是不支持线程公平的访问队列，这里的公平性指的是阻塞的线程可以按照先后顺序访问队列，即先阻塞的线程先访问。如果要保证线程公平访问，通常会降低吞吐量，当然 ArrayBlockingQueue 也是支持公平访问的，使用方式如下：

ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);

这里小伙伴可以思考下，公平锁是如何实现的？

之前我们的文章也讲过，可以使用可重入锁(对于可重入锁不太了解的伙伴可以看我这个系列专栏前面的文章)，具体代码如下：

public ArrayBlockingQueue(int capacity, boolean fair) {    if (capacity <= 0)        throw new IllegalArgumentException();    this.items = new Object[capacity];    lock = new ReentrantLock(fair);    notEmpty = lock.newCondition();    notFull =  lock.newCondition();}

ArrayBlockingQueue 源码解析

如果队列是空的，消费者会一直等待，当生产者添加元素时，消费者是如何知道当前队列有元素的呢？如果让你来设计阻塞队列你会如何设计，如何让生产者和消费者进行高效率的通信呢？让我们先来看看JDK是如何实现的。

入队操作

「使用通知模式实现」。所谓通知模式，就是当生产者往满的队列里添加元素时会阻塞住生产者，当消费者消费了一个队列中的元素后，会通知生产者当前队列可用。

通过查看JDK源码发现 ArrayBlockingQueue 使用了Condition来实现，代码如下：

public void put(E e) throws InterruptedException {    checkNotNull(e);    final ReentrantLock lock = this.lock;    // 使用condition，必须要获取锁    lock.lockInterruptibly();    try {        while (count == items.length)            // 队列满则在notFull对象上等待通知            notFull.await();        // 将指定元素放在队尾，并通知 notEmpty.signal();        enqueue(e);    } finally {        lock.unlock();    }}

上面的代码不难发现，当 items 这个数组满了的时候，进入 while 死循环在 notFull 这个 Condition 对象上等待，同时会释放掉可重入锁。看到这里我们会看到如果队列不满，则直接将当前元素插入到队尾，并且要通知因为队列为空而等待在 notEmpty 这个 condition 对象上的线程，具体如何通知呢？我们顺着源码继续往下看。

// 因为阻塞队列是线程安全的，只会获取到lock锁才能进入入队操作private void enqueue(E x) {    // assert lock.getHoldCount() == 1;    // assert items[putIndex] == null;    final Object[] items = this.items;    items[putIndex] = x;    if (++putIndex == items.length)        putIndex = 0;    // 队列的元素数量，插入后要增加1    count++;    // 因为插入了元素，所以要通知那些因为队列是空而等待的线程    notEmpty.signal();}

代码很简单就不做啰嗦了，其实就是入队时调用队列为空的 Condition 的 signal 方法。

出队操作

public E take() throws InterruptedException {    final ReentrantLock lock = this.lock;    lock.lockInterruptibly();    try {        // 队列为空在notEmpty条件进行阻塞        while (count == 0)            notEmpty.await();        // 出队，并且调用 notFull.signal(); 通知入队线程        return dequeue();    } finally {        lock.unlock();    }}

如果你看懂了入队操作的代码，那么出队这块其实是类似的，队列为空就阻塞，非空则取出首元素。

当然啦 dequeue() 方法在出队之后也会通知所有因为队列满而阻塞的入队线程，我们来看下代码：

// 出队操作，必须先获取到锁才能调用private E dequeue() {    // assert lock.getHoldCount() == 1;    // assert items[takeIndex] != null;    final Object[] items = this.items;    @SuppressWarnings("unchecked")    E x = (E) items[takeIndex];    // 出队一个元素，将这个位置置为null，即为删除    items[takeIndex] = null;    if (++takeIndex == items.length)        takeIndex = 0;    // 队列大小减一    count--;    if (itrs != null)        itrs.elementDequeued();    // 通知所有因为队列满阻塞的线程    notFull.signal();    return x;}

看到这里，那恭喜你你已经对阻塞队列的实现有了一个清晰的认识：它们是利用可重入锁获取两个 Condition 对象来分别阻塞入队和出队操作的。

你可能会问，阻塞队列使用了可重入锁，它是怎么来阻塞当前线程的？

从上面的代码，我们看到了当队列满调用了 notFull.await() 方法，这个方法是如何实现阻塞的呢？代码如下：

 public final void await() throws InterruptedException {     if (Thread.interrupted())         throw new InterruptedException();     // 添加一个新的节点线程到等待队列     Node node = addConditionWaiter();     int savedState = fullyRelease(node);     int interruptMode = 0;     while (!isOnSyncQueue(node)) {         // 阻塞主要通过 LockSupport.park(this)来实现         LockSupport.park(this);         if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)             break;     }     if (acquireQueued(node, savedState) && interruptMode != THROW_IE)         interruptMode = REINTERRUPT;     if (node.nextWaiter != null) // clean up if cancelled         unlinkCancelledWaiters();     if (interruptMode != 0)         reportInterruptAfterWait(interruptMode); }

将当前节点加入等待条件队列，通过调用 LockSupport.park(this) 方法来阻塞当前线程；

public static void park(Object blocker) {   Thread t = Thread.currentThread();   // 先保存一下将要阻塞的线程   setBlocker(t, blocker);   // unsafe.park阻塞 当前线程，这是个native方法   UNSAFE.park(false, 0L);   setBlocker(t, null);}

看到 UNSAFE.park() 方法，这是一个 native 方法，被 native修饰则表示是被 JVM 实现的，而 JVM 是在不同的操作系统中实现是不一样的，具体就是通过 C 来实现的，这块就不再深入了，因为我们是搞 Java 的知道这里已经很不错了，当然如果你很感兴趣可以自行在研究哈。

当调用了 UNSAFE.park() 方法就会阻塞当前线程，只有出现下面四种情况才会被唤醒从这个方法返回：

被中断时，这个方法是可以响应中断的；
等待一定的时间，超时返回，这个时间是 park() 方法的一个参数；
调用了 unpark() 方法；
异常现象的发生，原因不明，也就是意外情况；

动手实践

既然我们已经学习了 ArrayBlockingQueue ，知道了它是使用数组实现的，它是有界的，当队列满时会阻塞生产者线程，那么我们一起验证下:

public static void main(String[] args) throws InterruptedException {    ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(2);    arrayBlockingQueue.offer(1);    arrayBlockingQueue.put("2");    System.out.println("满了");    arrayBlockingQueue.put(3);    for (Object i : arrayBlockingQueue) {        System.out.println(i);    }}

运行上述代码，可以看到队列大小为2，当我们插入第3个元素时，就会阻塞当前线程，当前线程被阻塞后也就进入到了 WATING 状态，建议大家也动手验证下，实践出真知。

使用 jps 查看当前运行程序的进程号；
然后使用 jstack -l pid 查询线程堆栈信息；

原创真心不易，希望你能帮我个小忙呗，如果本文内容你觉得有所收获，请帮忙点个“在看”，或者转发分享让更多的小伙伴看到。

● 观察者模式解析

● Java并发|Atomic包下的原子操作类

● Java并发|Semaphore如何实现限流器

觉得写得还不错的小伙伴麻烦动手「三连」；
文章如果存在不正确的地方，麻烦指出，非常感谢您的阅读；
推荐大家关注我的公众号，会为你定期推送原创干货文章，拉你进优质学习社群；
github地址：https://github.com/coderluojust/qige_blogs