【JUC并发编程】ArrayBlockingQueue和LinkedBlockingQueue源码2分钟看完

1、BlockingQueue

BlockingQueue是JUC包下提供的一个阻塞队列 接口；

1）接口方法

队列操作

• 抛出异常：add(e)、remove()、element()
• 返回特定值：offer()队尾入队/poll()删除队头元素/peek()
• 一直阻塞：put(e)/take()
• 超时退出：offer(e,time,unit)/poll(time,unit)

其中：BlockingQueue 不接受 null 元素。试图 add 、 put 或 offer ⼀个 null 元素时，某些实现会抛出 NullPointerException 。

2）阻塞队列分类

• ArrayBlockingQueue：由数组结构组成的有界阻塞队列。

• LinkedBlockingQueue：由链表组成的有界阻塞队列。

• PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列(默认升序排序)。

• DelayQueue：支持延时获取元素的无界阻塞队列。

  队列使用PriorityQueue来实现。

• SynchronousQueue：一个不存储元素的阻塞队列。

• LinkedTransferQueue：由链表组成的无界阻塞队列。

  其多了tryTransfer()方法和transfer()方法。

  • transfer()：可以把生产者传入的元素立刻传输给消费者。如果没有consumer在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。
  • tryTransfer()：用来试探生产者传入的元素是否能直接传给消费者。不管是否有consumer正在等待接收元素，都立刻返回。

2、ArrayBlockingQueue

ArrayBlockingQueue是由数组结构组成的有界阻塞队列。通过ReentrantLock保证线程安全、并实现 Producer-Consumer模式。

1）构造函数

从构造函数可知，默认采用非公平锁；

public ArrayBlockingQueue(int capacity, boolean fair) {if (capacity <= 0)throw new IllegalArgumentException();// 底层使用对象数组保存元素this.items = new Object[capacity];// 初始化需要加锁使用的ReentrantLock实例，默认采用非公平锁lock = new ReentrantLock(fair);/*** 判断队列是 空 or 满*     notEmpty⽤于执⾏take时进⾏await()等待操作，put时进⾏signal()唤醒操作*     notFull⽤于执⾏take时进⾏signal()唤醒操作，put时进⾏await()等待操作*/notEmpty = lock.newCondition();notFull =  lock.newCondition();
}

方法释义：

• 构造函数的入参capacity指定了底层存储元素数组⻓度的⼤⼩；
• 初始化需要加锁使⽤的ReentrantLock实例，默认采⽤的是⾮公平锁；
• 基于Lock的Condition判断队列是 空 or 满
  • notEmpty⽤于执⾏take时进⾏await()等待操作，put时进⾏signal()唤醒操作；
  • notFull⽤于执⾏take时进⾏signal()唤醒操作，put时进⾏await()等待操作；

2）put()入队

public void put(E e) throws InterruptedException {// 入参不能为空checkNotNull(e);final ReentrantLock lock = this.lock;// 加可中断锁lock.lockInterruptibly();try {while (count == items.length)// 队列满了，则阻塞等待signal唤醒，同时释放次有的锁。notFull.await();// 入队操作enqueue(e);} finally {lock.unlock();}
}private void enqueue(E x) {final Object[] items = this.items;items[putIndex] = x;// 如果数据已经插入到数组末尾，则重置putIndex为0，从0开始继续插入。if (++putIndex == items.length)putIndex = 0;count++;// 通知take线程解除阻塞notEmpty.signal();
}

方法释义：

• ⾸先尝试获得可中断锁，即：lock.lockInterruptibly()，当执⾏interrupt操作时，该锁可以被中断。
• 如果数组中元素的个数（count）等于数组的⻓度了，说明队列已经满了，在该线程上执⾏等待操作：notFull.await()； ，等待signal唤醒。
• 如果队列没有满，则调⽤enqueue(e)⽅法执⾏⼊列操作；
  1. ⼊列操作⾸先会将待插⼊值x放⼊数组下标为putIndex的位置上，然后再将putIndex加1，来指向下⼀次插⼊的下标位置。
     • 如果加1后的putIndex等于了数组的⻓度，那么说明已经越界了（因为putIndex是从0开始的）；则做循环式插⼊，重置putIndex为0，从0开始继续插入。
• 最后，执⾏count++来计算元素总个数；并调⽤notEmpty.signal()⽅法来解除阻塞；
  • 当队列为空的时候，执⾏take⽅法会被notEmpty.await()阻塞；
  • 此处就是通知take线程解除阻塞。

3）take()出队

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)// 如果队列为空，则线程阻塞 等待signal唤醒，释放持有的锁notEmpty.await();// 执行出队操作return dequeue();} finally {lock.unlock();}
}private E dequeue() {final Object[] items = this.items;@SuppressWarnings("unchecked")// 获取takeIndex下标的元素E x = (E) items[takeIndex];// 将takeIndex下标下的袁术置为null，便于后面GC回收items[takeIndex] = null;// 如果出队操作的到了数组末尾，则重置takeIndex，从0开始继续取出if (++takeIndex == items.length)takeIndex = 0;count--;// 默认itrs为null，不会走进if代码段。if (itrs != null)itrs.elementDequeued();// 通知put线程解除阻塞notFull.signal();return x;
}

方法释义：

• take⽅法和put⽅法类似，区别是出队的指针是takeIndex；

• ⾸先尝试获得可中断锁，即：lock.lockInterruptibly()，当执⾏interrupt操作时，该锁可以被中断。

• 如果队列中为空；执⾏notEmpty.await()将线程阻塞 等待signal唤醒，释放持有的锁。

  • 当调⽤put⽅法向队列中放⼊元素之后 ，会调⽤notEmpty.signal⽅法对等待的线程执⾏唤醒操作；

• 如果出队操作的到了数组末尾，则重置takeIndex，从0开始继续取出；

• 出队执⾏完毕后，调⽤notFull.signal⽅法来唤醒在notFull上⾯

  await的线程。 通知put线程解除阻塞。

3、LinkedBlockingQueue

LinkedBlockingQueue是由链表结构组成的有界阻塞队列。通过ReentrantLock保证线程安全、并实现 Producer-Consumer模式。

1）构造函数

如果不指定容量，则默认LinkedBlockingQueue是无界阻塞队列（capacity = Integer.MAX_VALUE）

构造函数中会创建⼀个空的节点，作为整个链表的头节点。

2）put()入队

private void enqueue(Node<E> node) {// assert putLock.isHeldByCurrentThread();// assert last.next == null;last = last.next = node;
}

整个put()流程和ArrayBlockingQueue基本一致，对于链表容量的统计会额外采用一个AtomiceInteger类型的变量count维护。

• • 最后唤醒put()线程的代码段上有一个 c == 0的判断，这里的c是入队操作之前的数量。

/** Current number of elements */
private final AtomicInteger count = new AtomicInteger();/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

这里有个比较有意思的点：

• 入队操作添加完元素之后，如果发现当前队列的元素数量还没到最大容量，则尝试唤醒其他put()操作阻塞的线程；

if (c + 1 < capacity)notFull.signal();

3）take()出队

private E dequeue() {// assert takeLock.isHeldByCurrentThread();// assert head.item == null;Node<E> h = head;Node<E> first = h.next;h.next = h; // help GChead = first;E x = first.item;first.item = null;return x;
}

整个take()流程和ArrayBlockingQueue基本一致，稍微看一下即可。

• 最后唤醒take()线程的代码段上有一个 c == capacity的判断，这里的c是出队操作之前的数量。

和LinkedBlockingQueue的put()操作一样：

• 出队操作移除完元素之后，如果发现当前队列的元素数量 > 1，则尝试唤醒其他take()操作阻塞的线程；

if (c > 1)notEmpty.signal();