详解ArrayBlockQueue源码解析-eolink官网

详解ArrayBlockQueue源码解析

今天要讲的是ArrayBlockQueue，ArrayBlockQueue是JUC提供的线程安全的有界的阻塞队列，一看到Array，第一反应：这货肯定和数组有关，既然是数组，那自然是有界的了，我们先来看看ArrayBlockQueue的基本使用方法，然后再看看ArrayBlockQueue的源码。

ArrayBlockQueue基本使用

public static void main(String[] args) throws InterruptedException {

ArrayBlockingQueue arrayBlockingQueue=new ArrayBlockingQueue(5);

arrayBlockingQueue.offer(10);

arrayBlockingQueue.offer(50);

arrayBlockingQueue.add(20);

arrayBlockingQueue.add(60);

System.out.println(arrayBlockingQueue);

System.out.println(arrayBlockingQueue.poll());

System.out.println(arrayBlockingQueue);

System.out.println(arrayBlockingQueue.take());

System.out.println(arrayBlockingQueue);

System.out.println(arrayBlockingQueue.peek());

System.out.println(arrayBlockingQueue);

}

运行结果：

创建了一个长度为5的ArrayBlockQueue。

用offer方法，向ArrayBlockQueue添加了两个元素，分别是10，50。

用put方法，向ArrayBlockQueue添加了两个元素，分别是20，60。

打印出ArrayBlockQueue，结果是10，50，20，60。

用poll方法，弹出ArrayBlockQueue第一个元素，并且打印出来：10。

打印出ArrayBlockQueue，结果是50，http://20，60。

用take方法，弹出ArrayBlockQueue第一个元素，并且打印出来：50。

打印出ArrayBlockQueue，结果是20，60。

用peek方法，弹出ArrayBlockQueue第一个元素，并且打印出来：20。

打印出ArrayBlockQueue，结果是20，60。

代码比较简单，但是你肯定会有疑问

offer/add（在上面的代码中没有演示）/put都是往队列里面添加元素，区别是什么？

poll/take/peek都是弹出队列的元素，区别是什么？

底层代码是如何保证线程安全的？

数据保存在哪里？

要解决上面几个疑问，最好的办法当然是看下源码，通过亲自阅读源码所产生的印象远远要比看视频，看博客，死记硬背最后的结论要深刻的多。就算真的忘记了，只要再看看源码，瞬间可以回忆http://起来。

ArrayBlockQueue源码解析

构造方法

ArrayBlockQueue提供了三个构造方法，如下图所示：

ArrayBlockingQueue(int capacity)

public ArrayBlockingQueue(int capacity) {

this(capacity, false);

}

这是最常用的构造方法，传入capacity，capacity是容量的意思，也就是ArrayBlockingQueue的最大长度，方法内部直接调用了第二个构造方法，传入的第二个参数为false。

ArrayBlockingQueue(int capacity, boolean fair)

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();

}

这个构造方法接受两个参数，分别是capacity和fair，fair是boolean类型的，代表是公平锁，还是非公平锁，可以看出如果我们用第一个构造方法来创建ArrayBlockingQueue的话，采用的是非公平锁，因为公平锁会损失一定的性能，在没有充足的理由的情况下，是没有必要采用公平锁的。

方法内部做了几件事情：

创建Object类型的数组，容量为capacity，并且赋值给当前类对象的items。

创建排他锁。

创建条件变量notEmpty 。

创建条件变量notFull。

至于排他锁和两个条件变量是做什么用的，看到后面就明白了。

ArrayBlockingQueue(int capacity, boolean fair,Collection extends E> c)

public ArrayBlockingQueue(int capacity, boolean fair,

Collection extends E> c) {

//调用第二个构造方法，方法内部就是初始化数组，排他锁，两个条件变量

this(capacity, fair);

final ReentrantLock lock = this.lock;

lock.lock(); // 开启排他锁

try {

int i = 0;

try {

// 循环传入的集合，把集合中的元素赋值给items数组，其中i会自增

for (E e : c) {

checkNotNull(e);

items[i++] = e;

}

} catch (ArrayIndexOutOfBoundsException ex) {

throw new IllegalArgumentException();

}

SjtQM count = i;//把i赋值给count

//如果i==capacity，也就是到了最大容量，把0赋值给putIndex，否则把i赋值给putIndex

putIndex = (i == capacity) ? 0 : i;

} finally {

lock.unlock();//释放排他锁

}

调用第二个构造方法，方法内部就是初始化数组items，排他锁lock，以及两个条件变量。

开启排他锁。

循环传入的集合，将集合中的元素赋值给items数组，其中i会自增。

把i赋值给count。

如果i==capacity，说明到了最大的容量，就把0赋值给putIndex，否则把i赋值给putIndex。

在finally中释放排他锁。

看到这里，我们应该明白这个构造方法的作用是什么了，就是把传入的集合作为ArrayBlockingQueuede初始化数据，但是我们又会有一个新的疑问：count，putIndex 是做什么用的。

offer(E e)

public boolean offer(E e) {

checkNotNull(e);

final ReentrantLock lock = this.lock;

lock.lock();//开启排他锁

try {

if (count == items.length)//如果count==items.length，返回false

return false;

else {

enqueue(e);//入队

return true;//返回true

}

} finally {SjtQM

lock.unlock();//释放锁

}

开启排他锁。

如果count==items.length，也就是到了最大的容量，返回false。

如果count

释放排他锁。

看到这里，我们应该可以明白了，ArrayBlockQueue是如何保证线程安全的，还是利用了ReentrantLock排他锁，count就是用来保存数组的当前大小的。我们再来看看enqueue方法。

private void enqueue(E x) {

final Object[] items = this.items;

items[putIndex] = x;

if (++putIndex == items.length)

putIndex = 0;

count++;

notEmpty.signal();

}

这方法比较简单，在代码里面就不写注释了，做了如下的操作：

把x赋值给items[putIndex] 。

将putIndex进行自增，如果自增后的值 == items.length，把0赋值给putIndex 。

执行count++操作。

调用条件变量notEmpty的signal方法，说明在某个地方，必定调用了notEmpty的await方法，这里就是唤醒因为调用notEmpty的await方法而被阻塞的线程。

这里就解答了一个疑问：putIndex是做什么的，就是入队元素的下标。

add(E e)

public boolean add(E e) {

return super.add(e);

}

public boolean add(E e) {

if (offer(e))

return true;

else

throw new IllegalStateException("Queue full");

}

这个方法内部最终还是调用的offer方法。

put(E e)

public void put(E e) throws InterruptedException {

checkNotNull(e);

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();//开启响应中断的排他锁

try {

while (count == items.length)//如果队列满了，调用notFull的await

notFull.await();

enqueue(e);//入队

} finally {

lock.unlock();//释放排他锁

}

开启响应中断的排他锁，如果在获取锁的过程中，当前的线程被中断，会抛出异常。

如果队列满了，调用notFull的await方法，说明在某个地方，必定调用了notFull的signal方法来唤醒当前线程，这里用while循环是为了防止虚假唤醒。

执行入队操作。

释放排他锁。

可以看到put方法和 offer/add方法的区别了：

offer/add：如果队列满了，直接返回false。

put：如果队列满了，当前线程被阻塞，等待唤醒。

poll()

public E poll() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

return (count == 0) ? null : dequeue();

} finally {

lock.unlock();

}

开启排他锁。

如果count==0，直接返回null，否则执行dequeue出队操作。

释放排他锁。

我们来看dequeue方法：

private E dequeue() {

final Object[] items = this.items;

@SuppressWarnings("unchecked")

E x = (E) items[takeIndex];//获得元素的值

items[takeIndex] = null;//把null赋值给items[takeIndex]

if (++takeIndex == items.length)//如果takeIndex自增后的值== items.length，就把0赋值给takeIndex

takeIndex = 0;

count--;

if (itrs != null)

itrs.elementDequeued();

notFull.signal();//唤醒因为调用notFull的await方法而被阻塞的线程

return x;

}

获取元素的值，takeIndex保存的是出队的下标。

把null赋值给items[takeIndex]，也就是清空被弹出的元素。

如果takeIndex自增后的值== items.length，就把0赋值给takeIndex。

count--。

唤醒因为调用notFull的await方法而被阻塞的线程。

这里调用了notFull的signal方法来唤醒因为调用notFull的await方法而被阻塞的线程，那到底在哪里调用了notFull的await方法呢，还记不记得在put方法中调用了notFull的await方法，我们再看看：

while (count == items.length)

notFull.await();

当队列满了，就调用 notFull.await()来等待，在出队操作中，又调用了notFull.signal()来唤醒。

take()

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

while (count == 0)

notEmpty.await();

return dequeue();

} finally {

lock.unlock();

}

开启排他锁。

如果count==0，代表队列是空的，则调用notEmpty的await方法，用while循环是为了防止虚假唤醒。

执行出队操作。

释放排他锁。

这里调用了notEmpty的await方法，那么哪里调用了notEmpty的signal方法呢？在enqueue入队方法里。

我们可以看到take和poll的区别：

take：如果队列为空，会阻塞，直到被唤醒了。

poll：如果队列为空，直接返回null。

peek()

public E peek() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

return itemAt(takeIndex);

} finally {

lock.unlock();

}

final E itemAt(int i) {

return (E) items[i];

}

开启排他锁。

获得元素。

释放排他锁。

我们可以看到peek和poll/take的区别：

peek，只是获取元素，不会清空元素。

poll/take，获取并清空元素。

size()

public int size() {

SjtQM final ReentrantLock lock = this.lock;

lock.lock();

try {

return count;

} finally {

lock.unlock();

}

开启排他锁。

返回count。

释放排他锁。

总结

至此，ArrayBlockQueue的核心源码就分析完毕了，我们来做一个总结：

ArrayBlockQueue有几个比较重要的字段，分别是items，保存的是队列的数据，putIndex保存的是入队的下标，takeIndex保存的是出队的下标，count用来统计队列元素的个数，lock用来保证线程的安全性，notEmpty和notFull两个条件变量实现唤醒和阻塞。

offer和add是一样的，其中add方法内部调用的就是offer方法，如果队列满了，直接返回false。

put，如果队列满了，会被阻塞。

peek，只是弹出元素，不会清空元素。

poll，弹出并清空元素，如果队列为空，直接返回null。

take，弹出并清空元素，如果队列为空，会被阻塞。

以上所述是给大家介绍的ArrayBlockQueue源码解析详解整合，希望对大家有所帮助，如果大家有任何疑问请给我留言，会及时回复大家的。在此也非常感谢大家对我们网站的支持！

释放排他锁。

private void enqueue(E x) {

final Object[] items = this.items;

items[putIndex] = x;

if (++putIndex == items.length)

putIndex = 0;

count++;

notEmpty.signal();

}

这方法比较简单，在代码里面就不写注释了，做了如下的操作：

把x赋值给items[putIndex] 。

将putIndex进行自增，如果自增后的值 == items.length，把0赋值给putIndex 。

执行count++操作。

调用条件变量notEmpty的signal方法，说明在某个地方，必定调用了notEmpty的await方法，这里就是唤醒因为调用notEmpty的await方法而被阻塞的线程。

这里就解答了一个疑问：putIndex是做什么的，就是入队元素的下标。

add(E e)

public boolean add(E e) {

return super.add(e);

}

public boolean add(E e) {

if (offer(e))

return true;

else

throw new IllegalStateException("Queue full");

}

这个方法内部最终还是调用的offer方法。

put(E e)

public void put(E e) throws InterruptedException {

checkNotNull(e);

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();//开启响应中断的排他锁

try {

while (count == items.length)//如果队列满了，调用notFull的await

notFull.await();

enqueue(e);//入队

} finally {

lock.unlock();//释放排他锁

}

开启响应中断的排他锁，如果在获取锁的过程中，当前的线程被中断，会抛出异常。

如果队列满了，调用notFull的await方法，说明在某个地方，必定调用了notFull的signal方法来唤醒当前线程，这里用while循环是为了防止虚假唤醒。

执行入队操作。

释放排他锁。

可以看到put方法和 offer/add方法的区别了：

offer/add：如果队列满了，直接返回false。

put：如果队列满了，当前线程被阻塞，等待唤醒。

poll()

public E poll() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

return (count == 0) ? null : dequeue();

} finally {

lock.unlock();

}

开启排他锁。

如果count==0，直接返回null，否则执行dequeue出队操作。

释放排他锁。

我们来看dequeue方法：

private E dequeue() {

final Object[] items = this.items;

@SuppressWarnings("unchecked")

E x = (E) items[takeIndex];//获得元素的值

items[takeIndex] = null;//把null赋值给items[takeIndex]

if (++takeIndex == items.length)//如果takeIndex自增后的值== items.length，就把0赋值给takeIndex

takeIndex = 0;

count--;

if (itrs != null)

itrs.elementDequeued();

notFull.signal();//唤醒因为调用notFull的await方法而被阻塞的线程

return x;

}

获取元素的值，takeIndex保存的是出队的下标。

把null赋值给items[takeIndex]，也就是清空被弹出的元素。

如果takeIndex自增后的值== items.length，就把0赋值给takeIndex。

count--。

唤醒因为调用notFull的await方法而被阻塞的线程。

while (count == items.length)

notFull.await();

当队列满了，就调用 notFull.await()来等待，在出队操作中，又调用了notFull.signal()来唤醒。

take()

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

while (count == 0)

notEmpty.await();

return dequeue();

} finally {

lock.unlock();

}

开启排他锁。

如果count==0，代表队列是空的，则调用notEmpty的await方法，用while循环是为了防止虚假唤醒。

执行出队操作。

释放排他锁。

这里调用了notEmpty的await方法，那么哪里调用了notEmpty的signal方法呢？在enqueue入队方法里。

我们可以看到take和poll的区别：

take：如果队列为空，会阻塞，直到被唤醒了。

poll：如果队列为空，直接返回null。

peek()

public E peek() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

return itemAt(takeIndex);

} finally {

lock.unlock();

}

final E itemAt(int i) {

return (E) items[i];

}

开启排他锁。

获得元素。

释放排他锁。

我们可以看到peek和poll/take的区别：

peek，只是获取元素，不会清空元素。

poll/take，获取并清空元素。

size()

public int size() {

SjtQM final ReentrantLock lock = this.lock;

lock.lock();

try {

return count;

} finally {

lock.unlock();

}

开启排他锁。

返回count。

释放排他锁。

总结

至此，ArrayBlockQueue的核心源码就分析完毕了，我们来做一个总结：

offer和add是一样的，其中add方法内部调用的就是offer方法，如果队列满了，直接返回false。

put，如果队列满了，会被阻塞。

peek，只是弹出元素，不会清空元素。

poll，弹出并清空元素，如果队列为空，直接返回null。

take，弹出并清空元素，如果队列为空，会被阻塞。

Flask接口签名sign原理与实例代码浅析

209 2023-01-09

详解ArrayBlockQueue源码解析

Flask接口签名sign原理与实例代码浅析

java中的接口是类吗

vue项目接口域名动态的获取方法

推荐文章

接口调用是什么意思？几种常用接口调用方式

接口设计原则

8款在线 API 接口文档管理工具

api管理系统是什么？

什么是接口调试？接口调试的步骤有哪些？

api 接口管理系统有哪些？

接口测试有几种测试方法

API文档生成工具有哪些？

微服务和api网关区别

交换机配置步骤

最近发表

热评文章

在线接口文档管理工具推荐，支持在线测试，HTTP接口

开源的在线接口文档wiki工具Mindoc的介绍与使

如何优雅的进行接口设计？接口设计的六大原则是什么？

什么是API测试,api检测公司

软件接口设计怎么做？前后端分离软件接口设计思路

接口管理平台推荐，几大接口管理平台总有一款适合你！