ArrayBlockingQueue阻塞队列学习总结

队列学习总结"/>

ArrayBlockingQueue阻塞队列学习总结

文章目录

• 前言
• 一、ArrayBlockingQueue构造函数初始化数据
• 二、add、put、poll、peek、take、offer、remove等方法调用
• 三、Itrs类迭代器的使用
• 总结

前言

        ArrayBlockingQueue是一个基于数组的阻塞队列实现，在多线程环境下，自动管理了多线间的自动等待于唤醒功能，从而使得程序员可以忽略这些细节，关注更高级的功能。

一、ArrayBlockingQueue构造函数初始化数据

        初始构造函数内定义容量大小capacity，是否采用公平锁fair,以及源集合数据。自定义数组items，赋予容量大小长度并将集合数据导入items数组，定义putIndex为下一次插入队列的起点，若队列已满则为0。notEmpty和notFull分别为take和put操作的两个阻塞条件。

二、add、put、poll、peek、take、offer、remove等方法调用

1.add方法

该方法直接调offer，如果队列已满，直接throw new IllegalStateException("Queue full")

public boolean add(E e) {return super.add(e);}

2.put方法

加可中断锁并且在队列满的时候利用方法notFull.await造成线程阻塞，未满则调用enqueue(e)方法将元素插入，并且，激活一个notEmpty，即take阻塞条件的线程。

public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == items.length)notFull.await();enqueue(e);} finally {lock.unlock();}}

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;count++;notEmpty.signal();}

3.poll方法

加锁并且在队列数量为0时返回null，否则利用方法dequeue获取并移除此队列的头元素，并且，激活一个notFull，即put阻塞条件的线程。

 public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {return (count == 0) ? null : dequeue();} finally {lock.unlock();}}

private E dequeue() {// assert lock.getHoldCount() == 1;// assert items[takeIndex] != null;final Object[] items = this.items;@SuppressWarnings("unchecked")E x = (E) items[takeIndex];items[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)itrs.elementDequeued();notFull.signal();return x;}

4.peek方法

直接返回当前队列的头元素，但不删除

public E peek() {final ReentrantLock lock = this.lock;lock.lock();try {return itemAt(takeIndex); // null when queue is empty} finally {lock.unlock();}}

5.take方法

加可中断锁并且在队列空的时候利用方法notEmpty.await造成线程阻塞，未空则调用enqueue(e)方法将元素拿走，并且，激活一个notFull，即put阻塞条件的线程

public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)notEmpty.await();return dequeue();} finally {lock.unlock();}}

6.offer方法

只有集合元素时加常规锁并且在队列满的时候返回false，未满则调用enqueue(e)方法将元素插入，并且，激活一个notEmpty，即take阻塞条件的线程；不仅有集合元素还有时间参数时加可中断锁，并且在队列满的时候进入线程等待，等待时间到后激活线程如果依旧是满则返回false，未满则调用enqueue(e)方法将元素插入，并且，激活一个notEmpty，即take阻塞条件的线程。

public boolean offer(E e) {checkNotNull(e);final ReentrantLock lock = this.lock;lock.lock();try {if (count == items.length)return false;else {enqueue(e);return true;}} finally {lock.unlock();}}

public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {checkNotNull(e);long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == items.length) {if (nanos <= 0)return false;nanos = notFull.awaitNanos(nanos);}enqueue(e);return true;} finally {lock.unlock();}}

7.remove方法

删除元素，do-while循环找到元素并在remveAt方法中删除该索引所对应元素。在removeAt方法中，如果删除元素时队列头元素则直接删除，否则将元素删除后所有往后索引及所对应元素前移。

public boolean remove(Object o) {if (o == null) return false;final Object[] items = this.items;final ReentrantLock lock = this.lock;lock.lock();try {if (count > 0) {final int putIndex = this.putIndex;int i = takeIndex;do {if (o.equals(items[i])) {removeAt(i);return true;}if (++i == items.length)i = 0;} while (i != putIndex);}return false;} finally {lock.unlock();}}

void removeAt(final int removeIndex) {// assert lock.getHoldCount() == 1;// assert items[removeIndex] != null;// assert removeIndex >= 0 && removeIndex < items.length;final Object[] items = this.items;if (removeIndex == takeIndex) {// removing front item; just advanceitems[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)itrs.elementDequeued();} else {// an "interior" remove// slide over all others up through putIndex.final int putIndex = this.putIndex;for (int i = removeIndex;;) {int next = i + 1;if (next == items.length)next = 0;if (next != putIndex) {items[i] = items[next];i = next;} else {items[i] = null;this.putIndex = i;break;}}count--;if (itrs != null)itrs.removedAt(removeIndex);}notFull.signal();}

三、Itrs类迭代器的使用

 if (itrs != null)itrs.elementDequeued();

if (itrs != null)itrs.removedAt(removeIndex);

        在 ArrayBlockingQueue 中使用 Itrs 维护这一个 Itr 链表，用于在一个队列下的多个 Itr 迭代器中共享队列元素，保证多个迭代器中的元素数据的一致性。

总结

        阻塞队列在并发状态下能满足线程的安全性，并且在对队列进行操作时有自己的阻塞以及开启线程处理，替开发人员考虑了队列满或空所造成的一些异常判断，在生产消费者模式的开发中，非常好用。