ArrayBlockingQueue 源码学习

2024-08-30 05:58
文章标签 源码 学习 arrayblockingqueue

本文主要是介绍ArrayBlockingQueue 源码学习，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

基于 JDK 1.8.0

1. 概括

ArrayBlockingQueue 是一个先进先出（FIFO）的阻塞队列，底层是数组，队列长度在创建的时候确定不能修改。

使用场景：生产者消费者

2. 类定义

public class ArrayBlockingQueue<E> extends AbstractQueue<E>implements BlockingQueue<E>, java.io.Serializable

ArrayBlockingQueue 继承于 AbstractQueue，实现 BlockingQueue，Serializable 接口。

  • BlockingQueue 的介绍参考 BlockingQueue 源码学习

  • AbstractQueue 源码查看

    public abstract class AbstractQueue<E>extends AbstractCollection<E>implements Queue<E>

    抽象类 AbstractQueue 继承于 AbstractCollection，实现 Queue 接口。

    其中 AbstractQueue 实现了 Queue 的一些操作，在插入时不允许为 NULL；

    根据 AbstractQueue 源码知道，AbstractQueue 对 Queue 接口的方法的返回值进行的参数值的判断，如果添加失败或者没有查找到元素，就会抛出相关异常，这里就不再展示。

3. 成员变量

/**
* 队列中的元素，final 类型不可变
*/
final Object[] items;/** 下一次执行 take，poll，peek，remove 操作的位置 */
int takeIndex;/** 下一次执行 put，offer，add 操作的位置，在哪个位置进行这些操作 */
int putIndex;/** 队列中的元素个数 */
int count;/** 锁，所有请求都会先去获取这个锁 */
final ReentrantLock lock;/** 条件队列，用于通知 take 方法告知队列中有元素了 */
private final Condition notEmpty;/** 条件队列，用于通知 put() 方法有空间放元素了 */
private final Condition notFull;/**
* TODO 迭代器，后面再展开学习
* Shared state for currently active iterators, or null if there
* are known not to be any.  Allows queue operations to update
* iterator state.
*/
transient Itrs itrs = null;

根据成员变量的介绍可以得知：

  • 该队列用数组保持元素
  • 入队和出队用了两个指针标记，这两个标记的移动类似于在环形节点上面移动，如上图，下面详细分析。
  • 并发控制使用了 ReentrantLock ，并且使用两个 Condition 来控制队列的入队方法和出队方法的阻塞与继续运行

4. 构造方法


/**
* 仅指定容量大小，默认使用  非公平锁  控制并发请求
*/
public ArrayBlockingQueue(int capacity) {this(capacity, false);
}/**
* 生成指定容量大小的数组，容量必须大于 0
* 构造 锁 和 条件队列
*/
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();
}/**
* 将集合 c 中的元素添加到 ArrayBlockingQueue 中
* count = c.size()
* 若集合 c 的容量大小等于 capacity，将 putIndex 置为 0
*/ 
public ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c) {this(capacity, fair);final ReentrantLock lock = this.lock;lock.lock(); // Lock only for visibility, not mutual exclusiontry {int i = 0;try {for (E e : c) {checkNotNull(e);items[i++] = e;}} catch (ArrayIndexOutOfBoundsException ex) {throw new IllegalArgumentException();}count = i;putIndex = (i == capacity) ? 0 : i;} finally {lock.unlock();}
}

5. 成员方法

在这里插入图片描述

5.1 入队方法

5.1.1 boolean add(E e)

调用父类 add 方法，而父类 AbstractQueue add 方法实际调用的是 offer(E e) 方法

public boolean add(E e) {return super.add(e);
}// super.add
public boolean add(E e) {if (offer(e))return true;elsethrow new IllegalStateException("Queue full");
}

5.1.2 boolean offer(E e) 与 boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException

  • 加锁后判断容量大小，满了就不添加；没满执行共用的入队方法，入队成功返回 true
/**
* 插入成功返回 true，否则返回 false
* 插入元素为 NULL 返回空指针异常
*/
public boolean offer(E e) {checkNotNull(e); // 首先判断元素是否为 NULL，为 NULL 抛出异常final ReentrantLock lock = this.lock;lock.lock();try {if (count == items.length)return false; // 若数组满，返回 falseelse {enqueue(e); // 添加元素并返回 truereturn true;}} finally {lock.unlock();}
}/**
* 带超时时间的添加元素方法
* 当超时时间达到，直接返回 false
*/ 
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();}
}

内部共用的 入队 方法 enqueue()

在这里插入图片描述

  • 仅在同步方法块中调用
  • putIndex 的值一直在从 0 到 items.length 之间循环，添加元素进来就往后移动一位，移动到最后就又回到了开头，像是在环形数组上面移动
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(); // 通知等待 take 的线程，表示当前数组有元素可取
}

5.1.3 void put(E e) throws InterruptedException

/**
* 阻塞的插入方法
*/
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();}
}

入队方法总结：

  • 根据 BlockingQueue 源码可知，添加元素有 3 个方法，分别是 add，offer，put，其中 add 方法添加元素失败会抛出异常，官方推荐使用 offer，put 方法带阻塞功能。
  • enqueue 是内部共用的添加元素方法，其中 putIndex 的值一直都在从表头 -> 表尾 -> 表头循环，有点类似在环上面移动
  • ArrayBlockingQueue 的 add，offer，put 方法都是线程安全的

5.2 出队方法

5.2.1 E poll() 与 E poll(long timeout, TimeUnit unit) throws InterruptedException

public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {return (count == 0) ? null : dequeue(); // 若队列没有元素，返回 NULL;否则执行出队操作，将表头节点弹出并返回} finally {lock.unlock();}
}public E poll(long timeout, TimeUnit unit) throws InterruptedException {long nanos = unit.toNanos(timeout);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0) { // 当数组没有元素时进行等待if (nanos <= 0)return null; // 等待超时，返回 NULLnanos = notEmpty.awaitNanos(nanos); }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]; //保存 takeIndex 处的元素，并将此处位置置为空items[takeIndex] = null;if (++takeIndex == items.length) // takeIndex 往后移动一位，若 takeIndex 已经到数组末尾，那么让他回到开头takeIndex = 0;count--;  // 队列数组元素减 1if (itrs != null)itrs.elementDequeued();notFull.signal(); // 通知等待 notFull 变量可用的线程，队列将可以添加元素进来return x;
}

5.2.2 E take() throws InterruptedException

  • 与 poll 不同的是，take 取元素会阻塞
public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)notEmpty.await();  // 没有元素时进行等待return dequeue(); // 否则取出表头节点} finally {lock.unlock();}
}

5.3 获取表头元素

  • 直接取 takeIndex 处的元素，队列为空时返回 NULL
public E peek() {final ReentrantLock lock = this.lock;lock.lock();try {return itemAt(takeIndex); // null when queue is empty} finally {lock.unlock();}
}

5.4 删除指定元素

在这里插入图片描述

  • 首先这个方法一定是在同步方法块中进行移除操作 ，环形图继续拿出来，假设原来是顺时针方向进行插入元素
  • 删除元素，就得从 takeIndex （表头位置）开始遍历，如果有相同的就移除当前位置的元素
  • 若 takeIndex 走到末尾，然后再置为 0，就像上面环一样，一直顺时针走
  • 若 takeIndex = putIndex ，说明走到遍历到末尾了，注意这里的末尾是插入位置 putIndex 的前一个位置，说明前面的位置已经遍历完了。
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 到表头i = 0;} while (i != putIndex); // i == putIndex 时，说明数组中元素已经全部遍历，因为 putIndex 是下一次插入元素的位置}return false;  // 队列为空，删除不了元素，返回 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;// 若待删除元素位置等于此时表头位置，直接将此处元素置为 NULL，并且 takeIndex 指向下一个位置，元素无需移动（参考上面环形图，移除 takeIndex 元素只需要置 NULL 然后后移）if (removeIndex == takeIndex) {// 元素值直接置为 nullitems[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;// 待删除位置已经是最后一个元素，将 next 置为 0if (next == items.length)next = 0;if (next != putIndex) {// 逆时针移动，把后面的元素移到前面来items[i] = items[next];i = next;} else {// 此处位置的下一个位置是 putIndex，那么把此处位置置为 nullitems[i] = null;this.putIndex = i;break;}}// 更新数组元素数量count--;if (itrs != null)// 更新迭代器itrs.removedAt(removeIndex);}notFull.signal(); // 唤醒等待的 put 线程
}void elementDequeued() {// assert lock.getHoldCount() == 1;if (count == 0)queueIsEmpty();else if (takeIndex == 0)takeIndexWrapped();
}

内部类( TODO 后面再补充)

class Itrs {}private class Itr implements Iterator<E> {}

这篇关于ArrayBlockingQueue 源码学习的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1119983。

相关文章

HarmonyOS学习(七)——UI（五）常用布局总结

HarmonyOS学习(七)——UI（五）常用布局总结

自适应布局 1.1、线性布局（LinearLayout） 通过线性容器Row和Column实现线性布局。Column容器内的子组件按照垂直方向排列，Row组件中的子组件按照水平方向排列。 属性说明space通过space参数设置主轴上子组件的间距，达到各子组件在排列上的等间距效果alignItems设置子组件在交叉轴上的对齐方式，且在各类尺寸屏幕上表现一致，其中交叉轴为垂直时，取值为Vert
阅读更多...
Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya（不是本人，claude AI）在社交媒体上分享了他在OpenAI学习到的15个Prompt撰写技巧。 以下是详细的内容： 提示精确化:在编写提示时,力求表达清晰准确。清楚地阐述任务需求和概念定义至关重要。例:不用"分析文本",而用"判断这段话的情感倾向:积极、消极还是中性"。 快速迭代:善于快速连续调整提示。熟练的提示工程师能够灵活地进行多轮优化。例:从"总结文章"到"用
阅读更多...
【前端学习】AntV G6-08 深入图形与图形分组、自定义节点、节点动画（下）

【前端学习】AntV G6-08 深入图形与图形分组、自定义节点、节点动画（下）

【课程链接】 AntV G6：深入图形与图形分组、自定义节点、节点动画（下）_哔哩哔哩_bilibili 本章十吾老师讲解了一个复杂的自定义节点中，应该怎样去计算和绘制图形，如何给一个图形制作不间断的动画，以及在鼠标事件之后产生动画。（有点难，需要好好理解） <!DOCTYPE html><html><head><meta charset="UTF-8"><title>06
阅读更多...
学习hash总结

学习hash总结

2014/1/29/   最近刚开始学hash，名字很陌生，但是hash的思想却很熟悉，以前早就做过此类的题，但是不知道这就是hash思想而已，说白了hash就是一个映射，往往灵活利用数组的下标来实现算法，hash的作用：1、判重；2、统计次数；
阅读更多...
JAVA智听未来一站式有声阅读平台听书系统小程序源码

JAVA智听未来一站式有声阅读平台听书系统小程序源码

智听未来，一站式有声阅读平台听书系统 🌟&nbsp;开篇：遇见未来，从“智听”开始 在这个快节奏的时代，你是否渴望在忙碌的间隙，找到一片属于自己的宁静角落？是否梦想着能随时随地，沉浸在知识的海洋，或是故事的奇幻世界里？今天，就让我带你一起探索“智听未来”——这一站式有声阅读平台听书系统，它正悄悄改变着我们的阅读方式，让未来触手可及！ 📚&nbsp;第一站：海量资源，应有尽有 走进“智听
阅读更多...
零基础学习Redis(10) -- zset类型命令使用

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合，内部除了存储元素外，还会存储一个score，存储在zset中的元素会按照score的大小升序排列，不同元素的score可以重复，score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]
阅读更多...
【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

【机器学习】高斯过程的基本概念和应用领域以及在python中的实例

引言 高斯过程（Gaussian Process，简称GP）是一种概率模型，用于描述一组随机变量的联合概率分布，其中任何一个有限维度的子集都具有高斯分布 文章目录 引言一、高斯过程1.1 基本定义1.1.1 随机过程1.1.2 高斯分布 1.2 高斯过程的特性1.2.1 联合高斯性1.2.2 均值函数1.2.3 协方差函数（或核函数） 1.3 核函数1.4 高斯过程回归（Gauss
阅读更多...
【学习笔记】 陈强-机器学习-Python-Ch15 人工神经网络（1）sklearn

【学习笔记】 陈强-机器学习-Python-Ch15 人工神经网络（1）sklearn

系列文章目录 监督学习：参数方法 【学习笔记】 陈强-机器学习-Python-Ch4 线性回归 【学习笔记】 陈强-机器学习-Python-Ch5 逻辑回归 【课后题练习】 陈强-机器学习-Python-Ch5 逻辑回归（SAheart.csv） 【学习笔记】 陈强-机器学习-Python-Ch6 多项逻辑回归 【学习笔记 及 课后题练习】 陈强-机器学习-Python-Ch7 判别分析 【学
阅读更多...
Java ArrayList扩容机制 （源码解读）

Java ArrayList扩容机制 （源码解读）

结论：初始长度为10，若所需长度小于1.5倍原长度，则按照1.5倍扩容。若不够用则按照所需长度扩容。 一. 明确类内部重要变量含义         1：数组默认长度         2:这是一个共享的空数组实例，用于明确创建长度为0时的ArrayList ，比如通过 new ArrayList<>(0)，ArrayList 内部的数组 elementData 会指向这个 EMPTY_EL
阅读更多...
系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

系统架构师考试学习笔记第三篇——架构设计高级知识（20）通信系统架构设计理论与实践

本章知识考点：         第20课时主要学习通信系统架构设计的理论和工作中的实践。根据新版考试大纲,本课时知识点会涉及案例分析题(25分),而在历年考试中,案例题对该部分内容的考查并不多,虽在综合知识选择题目中经常考查,但分值也不高。本课时内容侧重于对知识点的记忆和理解,按照以往的出题规律,通信系统架构设计基础知识点多来源于教材内的基础网络设备、网络架构和教材外最新时事热点技术。本课时知识
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2